Skip to content

Commit 325f8af

Browse files
committed
[VL] feat(iceberg): Add native orc write support
1 parent f734ed7 commit 325f8af

6 files changed

Lines changed: 56 additions & 3 deletions

File tree

backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,29 @@ class VeloxIcebergSuite extends IcebergSuite {
6363
}
6464
}
6565

66+
test("iceberg native ORC write") {
67+
withTable("iceberg_orc_write") {
68+
spark.sql("""
69+
|create table iceberg_orc_write(a int, b string) using iceberg
70+
|tblproperties ('write.format.default' = 'orc')
71+
|""".stripMargin)
72+
73+
val df = spark.sql("insert into iceberg_orc_write values (1, 'a'), (2, 'b')")
74+
assert(
75+
df.queryExecution.executedPlan
76+
.asInstanceOf[CommandResultExec]
77+
.commandPhysicalPlan
78+
.isInstanceOf[VeloxIcebergAppendDataExec])
79+
80+
checkAnswer(
81+
spark.sql("select * from iceberg_orc_write order by a"),
82+
Seq(Row(1, "a"), Row(2, "b")))
83+
checkAnswer(
84+
spark.sql("select distinct file_format from default.iceberg_orc_write.files"),
85+
Seq(Row("ORC")))
86+
}
87+
}
88+
6689
test("iceberg insert partition table identity transform") {
6790
withTable("iceberg_tb2") {
6891
spark.sql("""

cpp/velox/compute/VeloxBackend.cc

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040
#endif
4141

4242
#include "compute/VeloxRuntime.h"
43+
#include "compute/iceberg/IcebergWriter.h"
4344
#include "config/VeloxConfig.h"
4445
#ifdef ENABLE_S3
4546
#include "filesystem/GlutenS3FileSystem.h"
@@ -246,6 +247,7 @@ void VeloxBackend::init(
246247
velox::parquet::registerParquetReaderFactory();
247248
velox::parquet::registerParquetWriterFactory();
248249
velox::orc::registerOrcReaderFactory();
250+
registerIcebergOrcWriterFactory();
249251
velox::exec::ExprToSubfieldFilterParser::registerParser(std::make_unique<SparkExprToSubfieldFilterParser>());
250252
velox::connector::hive::BufferedInputBuilder::registerBuilder(std::make_shared<GlutenBufferedInputBuilder>());
251253

cpp/velox/compute/iceberg/IcebergWriter.cc

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,12 +26,34 @@
2626
#include "utils/ConfigExtractor.h"
2727
#include "velox/connectors/hive/iceberg/IcebergDataSink.h"
2828
#include "velox/connectors/hive/iceberg/IcebergDeleteFile.h"
29+
#include "velox/dwio/common/WriterFactory.h"
30+
#include "velox/dwio/dwrf/writer/Writer.h"
2931

3032
using namespace facebook::velox;
3133
using namespace facebook::velox::connector::hive;
3234
using namespace facebook::velox::connector::hive::iceberg;
3335
namespace {
3436

37+
class IcebergOrcWriterFactory final : public dwio::common::WriterFactory {
38+
public:
39+
IcebergOrcWriterFactory() : WriterFactory(dwio::common::FileFormat::ORC) {}
40+
41+
std::unique_ptr<dwio::common::Writer> createWriter(
42+
std::unique_ptr<dwio::common::FileSink> sink,
43+
const std::shared_ptr<dwio::common::WriterOptions>& options) override {
44+
auto orcOptions = std::dynamic_pointer_cast<dwrf::WriterOptions>(options);
45+
VELOX_CHECK_NOT_NULL(orcOptions, "Iceberg ORC writer expected DWRF writer options.");
46+
VELOX_CHECK_EQ(orcOptions->format, dwrf::DwrfFormat::kOrc);
47+
return std::make_unique<dwrf::Writer>(std::move(sink), *orcOptions);
48+
}
49+
50+
std::unique_ptr<dwio::common::WriterOptions> createWriterOptions() override {
51+
auto options = std::make_unique<dwrf::WriterOptions>();
52+
options->format = dwrf::DwrfFormat::kOrc;
53+
return options;
54+
}
55+
};
56+
3557
// Custom Iceberg file name generator for Gluten
3658
class GlutenIcebergFileNameGenerator : public connector::hive::FileNameGenerator {
3759
public:
@@ -175,6 +197,10 @@ std::shared_ptr<IcebergInsertTableHandle> createIcebergInsertTableHandle(
175197
} // namespace
176198

177199
namespace gluten {
200+
void registerIcebergOrcWriterFactory() {
201+
dwio::common::registerWriterFactory(std::make_shared<IcebergOrcWriterFactory>());
202+
}
203+
178204
IcebergWriter::IcebergWriter(
179205
const RowTypePtr& rowType,
180206
int32_t format,

cpp/velox/compute/iceberg/IcebergWriter.h

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@
2525

2626
namespace gluten {
2727

28+
void registerIcebergOrcWriterFactory();
29+
2830
struct WriteStats {
2931
uint64_t numWrittenBytes{0};
3032
uint32_t numWrittenFiles{0};

docs/get-started/VeloxIceberg.md

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -160,7 +160,7 @@ The "Gluten Support" column is now ready to be populated with:
160160

161161
| Spark option | Default | Description | Gluten Support |
162162
| --- | --- | --- | --- |
163-
| write-format | Table write.format.default | File format to use for this write operation; parquet, avro, or orc |⚠️ Parquet only|
163+
| write-format | Table write.format.default | File format to use for this write operation; parquet, avro, or orc |⚠️ Parquet and ORC|
164164
| target-file-size-bytes | As per table property | Overrides this table's write.target-file-size-bytes | |
165165
| check-nullability | true | Sets the nullable check on fields | |
166166
| snapshot-property.custom-key | null | Adds an entry with custom-key and corresponding value in the snapshot summary (the snapshot-property. prefix is only required for DSv2) | |
@@ -194,7 +194,7 @@ extracted from https://iceberg.apache.org/docs/latest/configuration/
194194

195195
| Property | Default | Description | Gluten Support |
196196
| --- | --- | --- | --- |
197-
| write.format.default | parquet | Default file format for the table; parquet, avro, or orc | |
197+
| write.format.default | parquet | Default file format for the table; parquet, avro, or orc |⚠️ Parquet and ORC|
198198
| write.delete.format.default | data file format | Default delete file format for the table; parquet, avro, or orc | |
199199
| write.parquet.row-group-size-bytes | 134217728 (128 MB) | Parquet row group size | |
200200
| write.parquet.page-size-bytes | 1048576 (1 MB) | Parquet page size ||

gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergWriteExec.scala

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -108,7 +108,7 @@ trait IcebergWriteExec extends ColumnarV2TableWriteExec {
108108
return ValidationResult.failed("Not support write table with sort order")
109109
}
110110
val format = IcebergWriteUtil.getFileFormat(write)
111-
if (format != FileFormat.PARQUET) {
111+
if (!Seq(FileFormat.PARQUET, FileFormat.ORC).contains(format)) {
112112
return ValidationResult.failed("Not support this format " + format.name())
113113
}
114114

0 commit comments

Comments
 (0)