This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 8d2bd1e14b6 [fix](iceberg) Publish Parquet split offsets in data file
metadata (#69615)
8d2bd1e14b6 is described below
commit 8d2bd1e14b6d95d923d43e00857da326639cc466
Author: Gabriel <[email protected]>
AuthorDate: Mon Oct 26 21:44:57 2026 +0800
[fix](iceberg) Publish Parquet split offsets in data file metadata (#69615)
### What problem does this PR solve?
Doris writes Parquet row groups for Iceberg data files but does not
publish their split offsets in commit fragments or manifests. Readers
cannot use those known boundaries when planning splits.
Collect row-group starts from the closed writer's in-memory Footer,
including dictionary pages, and carry them through an optional Thrift
field into Iceberg data-file metadata. Keep collection independent of
column metrics policy; leave older BE fragments, empty files, ORC and
delete files without Parquet offsets.
When a file metadata-table scan requests `split_offsets`, also project
`file_size_in_bytes`: Iceberg uses the size to validate offsets and
otherwise returns null for valid values. Append this dependency after
the requested columns to preserve the BE JNI reader's positional
contract.
### Release note
Publish Parquet row-group split offsets in Iceberg manifests and
preserve them in projected file metadata-table scans.
### Validation
- Maven reactor build passed: 212 `IcebergScanPlanProviderTest` tests
and 31 `IcebergWriterHelperTest` tests, including Manifest round trips
and per-file isolation.
- The projection regression failed before the fix and passes afterward.
It covers files/data_files/all_files/all_data_files,
partitioned/unpartitioned tables and five requested-column orders.
- 6 targeted BE ASAN tests passed for dictionary/plain pages,
multiple/single/empty row groups, metrics-disabled Parquet commits and
unchanged ORC behavior. Changed source, generated Thrift and unity
objects were linked with cached engine libraries; this was not a clean
full BE rebuild.
- FE Checkstyle, clang-format 16, BE build hygiene and connector import
checks passed.
- Checked the optional Thrift field ID against master and release
branches 3.1, 4.0, 4.1 and 4.2.
- Extended the partitioned-write external regression to assert split
offsets, including with column metrics disabled. CI validation of the
latest revision and Spark/Trino interoperability remain pending.
### Check List (For Author)
- Test: Unit Test and Regression test
- Behavior changed: Yes. Newly written Iceberg Parquet data files
publish optional split offsets, which projected file metadata scans can
return.
- Does this need documentation? No.
---
.../writer/iceberg/viceberg_partition_writer.cpp | 10 ++++
.../format/transformer/viceberg_parquet_writer.cpp | 26 +++++++++
.../format/transformer/viceberg_parquet_writer.h | 2 +
.../iceberg/iceberg_partition_writer_test.cpp | 65 ++++++++++++++++++++-
.../format/transformer/vparquet_writer_test.cpp | 66 ++++++++++++++++++++++
.../connector/iceberg/IcebergScanPlanProvider.java | 9 ++-
.../connector/iceberg/IcebergWriterHelper.java | 8 ++-
.../iceberg/IcebergScanPlanProviderTest.java | 46 +++++++++++++++
.../connector/iceberg/IcebergWriterHelperTest.java | 54 ++++++++++++++++++
gensrc/thrift/DataSinks.thrift | 2 +
...est_iceberg_write_partition_distribution.groovy | 14 ++++-
11 files changed, 295 insertions(+), 7 deletions(-)
diff --git a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
index 5410b67a673..363776dcf17 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
+++ b/be/src/exec/sink/writer/iceberg/viceberg_partition_writer.cpp
@@ -217,6 +217,16 @@ Status
VIcebergPartitionWriter::_build_iceberg_commit_data(TIcebergCommitData* c
commit_data->__set_file_size(_file_format_transformer->written_len());
commit_data->__set_file_content(TFileContent::DATA);
commit_data->__set_partition_values(_partition_values);
+ // Split offsets describe the physical file, even when the table disables
all column metrics.
+ if (_file_format_type == TFileFormatType::FORMAT_PARQUET) {
+ std::vector<int64_t> offsets;
+
RETURN_IF_ERROR(static_cast<VIcebergParquetWriter*>(_file_format_transformer.get())
+ ->collect_split_offsets_after_close(&offsets));
+ // Empty files have no splits; leave the optional field absent, as
with an older BE.
+ if (!offsets.empty()) {
+ commit_data->__set_split_offsets(offsets);
+ }
+ }
// ORC collection reopens the file, so honor the FE policy before any
footer work.
if (!_collect_column_stats) {
return Status::OK();
diff --git a/be/src/format/transformer/viceberg_parquet_writer.cpp
b/be/src/format/transformer/viceberg_parquet_writer.cpp
index 5b951ec77b1..aa9ecb804cd 100644
--- a/be/src/format/transformer/viceberg_parquet_writer.cpp
+++ b/be/src/format/transformer/viceberg_parquet_writer.cpp
@@ -115,6 +115,32 @@ std::unique_ptr<ArrowBlockConvertor>
VIcebergParquetWriter::_create_arrow_block_
_iceberg_schema, &_iceberg_schema_json, timezone_name, timezone);
}
+Status
VIcebergParquetWriter::collect_split_offsets_after_close(std::vector<int64_t>*
offsets) {
+ auto metadata = _file_metadata();
+ // Arrow publishes file metadata only after Close(), so offsets cannot be
collected earlier.
+ if (metadata == nullptr) {
+ return Status::InternalError("File metadata is not available");
+ }
+ offsets->clear();
+ offsets->reserve(metadata->num_row_groups());
+ for (int i = 0; i < metadata->num_row_groups(); ++i) {
+ auto row_group = metadata->RowGroup(i);
+ // Empty files/groups have no page boundary to advertise as a scan
split.
+ if (row_group->num_rows() == 0 || row_group->num_columns() == 0) {
+ continue;
+ }
+ auto first_column = row_group->ColumnChunk(0);
+ int64_t offset = first_column->data_page_offset();
+ // A split must include the dictionary page, not just the first data
page that references it.
+ if (first_column->has_dictionary_page()) {
+ offset = std::min(offset, first_column->dictionary_page_offset());
+ }
+ offsets->push_back(offset);
+ }
+ std::sort(offsets->begin(), offsets->end());
+ return Status::OK();
+}
+
Status
VIcebergParquetWriter::collect_file_statistics_after_close(TIcebergColumnStats*
stats) {
std::shared_ptr<::parquet::FileMetaData> file_metadata = _file_metadata();
if (file_metadata == nullptr) {
diff --git a/be/src/format/transformer/viceberg_parquet_writer.h
b/be/src/format/transformer/viceberg_parquet_writer.h
index 787585eb569..f985371959b 100644
--- a/be/src/format/transformer/viceberg_parquet_writer.h
+++ b/be/src/format/transformer/viceberg_parquet_writer.h
@@ -44,6 +44,8 @@ public:
Status collect_file_statistics_after_close(TIcebergColumnStats* stats);
+ Status collect_split_offsets_after_close(std::vector<int64_t>* offsets);
+
protected:
std::unique_ptr<ArrowBlockConvertor> _create_arrow_block_convertor(
DataTypes types, std::vector<std::string> names, const
std::string& timezone_name,
diff --git a/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp
b/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp
index 9b10ff0e909..590e1953c67 100644
--- a/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp
+++ b/be/test/exec/sink/writer/iceberg/iceberg_partition_writer_test.cpp
@@ -15,16 +15,24 @@
// specific language governing permissions and limitations
// under the License.
+#include <arrow/memory_pool.h>
#include <gtest/gtest.h>
#include <optional>
+#include "core/data_type/data_type_number.h"
#include "exec/sink/writer/iceberg/viceberg_partition_writer.h"
#include "exec/sink/writer/iceberg/viceberg_sort_writer.h"
+#include "format/table/iceberg/schema_parser.h"
+#include "format/transformer/viceberg_parquet_writer.h"
#include "io/fs/local_file_system.h"
+#include "runtime/exec_env.h"
#include "testutil/mock/mock_descriptors.h"
#include "testutil/mock/mock_runtime_state.h"
#include "testutil/mock/mock_slot_ref.h"
+#include "util/defer_op.h"
+#include "util/timezone_utils.h"
+#include "util/uid_util.h"
namespace doris {
@@ -73,13 +81,14 @@ protected:
static std::unique_ptr<VIcebergPartitionWriter> make_writer(
const TDataSink& sink, const VExprContextSPtrs& output_exprs,
const iceberg::Schema& schema, const std::string* schema_json,
- const std::map<std::string, std::string>& hadoop_conf) {
+ const std::map<std::string, std::string>& hadoop_conf,
+ TFileFormatType::type format = TFileFormatType::FORMAT_ORC) {
IPartitionWriterBase::WriteInfo write_info;
write_info.file_type = TFileType::FILE_LOCAL;
return std::make_unique<VIcebergPartitionWriter>(
sink, std::vector<std::string> {}, output_exprs, schema,
schema_json,
- std::vector<std::string> {}, std::move(write_info), "data", 0,
- TFileFormatType::FORMAT_ORC, TFileCompressType::ZLIB,
hadoop_conf);
+ std::vector<std::string> {}, std::move(write_info), "data", 0,
format,
+ TFileCompressType::ZLIB, hadoop_conf);
}
static void install_fake_transformer(VIcebergPartitionWriter* writer,
@@ -90,6 +99,11 @@ protected:
output_exprs, std::move(close_status), destroyed);
}
+ static void install_parquet_transformer(VIcebergPartitionWriter* writer,
+
std::unique_ptr<VIcebergParquetWriter> transformer) {
+ writer->_file_format_transformer = std::move(transformer);
+ }
+
static Status build_commit_data(VIcebergPartitionWriter* writer,
TIcebergCommitData* commit_data) {
return writer->_build_iceberg_commit_data(commit_data);
@@ -114,6 +128,51 @@ TEST_F(VIcebergPartitionWriterTest,
OrcSkipsFooterCollectionWhenMetricsAreDisabl
EXPECT_FALSE(commit_data.__isset.column_stats);
}
+TEST_F(VIcebergPartitionWriterTest,
ParquetKeepsSplitOffsetsWhenColumnMetricsAreDisabled) {
+ TimezoneUtils::load_timezones_to_cache();
+ auto* env = ExecEnv::GetInstance();
+ auto* previous_pool = env->_arrow_memory_pool;
+ env->_arrow_memory_pool = arrow::default_memory_pool();
+ const auto path = "./iceberg_split_offsets_" +
UniqueId::gen_uid().to_string() + ".parquet";
+ auto fs = io::global_local_filesystem();
+ Defer cleanup([&] {
+ env->_arrow_memory_pool = previous_pool;
+ static_cast<void>(fs->delete_file(path));
+ });
+ auto type = std::make_shared<DataTypeInt32>();
+ auto output_exprs = MockSlotRef::create_mock_contexts(DataTypes {type});
+ const std::string json =
+
R"({"type":"struct","fields":[{"id":1,"name":"value","required":true,"type":"int"}]})";
+ auto schema = iceberg::SchemaParser::from_json(json);
+ io::FileWriterPtr file_writer;
+ ASSERT_TRUE(fs->create_file(path, &file_writer).ok());
+ RuntimeState state;
+ state.set_timezone("UTC");
+ ParquetFileOptions options {.compression_type =
TParquetCompressionType::SNAPPY,
+ .parquet_version =
TParquetVersion::PARQUET_1_0,
+ .parquet_disable_dictionary = false,
+ .enable_int96_timestamps = false};
+ auto transformer = std::make_unique<VIcebergParquetWriter>(
+ &state, file_writer.get(), output_exprs, std::vector<std::string>
{"value"}, false,
+ options, &json, *schema);
+ ASSERT_TRUE(transformer->open().ok());
+ auto column = type->create_column();
+ column->insert(Field::create_field<TYPE_INT>(7));
+ Block block;
+ block.insert({std::move(column), type, "value"});
+ ASSERT_TRUE(transformer->write(block).ok());
+ ASSERT_TRUE(transformer->close().ok());
+ auto writer = make_writer(make_table_sink(false), output_exprs, *schema,
&json, {},
+ TFileFormatType::FORMAT_PARQUET);
+ install_parquet_transformer(writer.get(), std::move(transformer));
+ TIcebergCommitData commit;
+ ASSERT_TRUE(build_commit_data(writer.get(), &commit).ok());
+ EXPECT_FALSE(commit.__isset.column_stats);
+ ASSERT_TRUE(commit.__isset.split_offsets);
+ ASSERT_EQ(1, commit.split_offsets.size());
+ EXPECT_EQ(4, commit.split_offsets[0]);
+}
+
TEST_F(VIcebergPartitionWriterTest,
MissingPolicyKeepsCollectionEnabledForRollingUpgrade) {
VExprContextSPtrs output_exprs;
iceberg::Schema schema(std::vector<iceberg::NestedField> {});
diff --git a/be/test/format/transformer/vparquet_writer_test.cpp
b/be/test/format/transformer/vparquet_writer_test.cpp
index acab031b796..6591ca24f57 100644
--- a/be/test/format/transformer/vparquet_writer_test.cpp
+++ b/be/test/format/transformer/vparquet_writer_test.cpp
@@ -28,6 +28,7 @@
#include <optional>
#include <string_view>
+#include "common/config.h"
#include "core/block/block.h"
#include "core/column/column_array.h"
#include "core/column/column_decimal.h"
@@ -50,6 +51,7 @@
#include "runtime/exec_env.h"
#include "runtime/runtime_state.h"
#include "testutil/mock/mock_slot_ref.h"
+#include "util/defer_op.h"
#include "util/timezone_utils.h"
#include "util/uid_util.h"
@@ -128,11 +130,75 @@ protected:
}
}
+ void check_split_offsets(int row_groups, bool dictionary) {
+ const auto previous_row_group_size = config::min_row_group_size;
+ config::min_row_group_size = 0;
+ Defer restore_row_group_size([&] { config::min_row_group_size =
previous_row_group_size; });
+ auto type = std::make_shared<DataTypeInt32>();
+ auto output_exprs = MockSlotRef::create_mock_contexts(DataTypes
{type});
+ const std::string json =
+
R"({"type":"struct","fields":[{"id":1,"name":"value","required":true,"type":"int"}]})";
+ auto schema = iceberg::SchemaParser::from_json(json);
+ io::FileWriterPtr file_writer;
+ ASSERT_TRUE(_fs->create_file(_file_path, &file_writer).ok());
+ RuntimeState state;
+ state.set_timezone("UTC");
+ ParquetFileOptions options {.compression_type =
TParquetCompressionType::SNAPPY,
+ .parquet_version =
TParquetVersion::PARQUET_1_0,
+ .parquet_disable_dictionary = !dictionary,
+ .enable_int96_timestamps = false};
+ VIcebergParquetWriter writer(&state, file_writer.get(), output_exprs,
{"value"}, false,
+ options, &json, *schema);
+ ASSERT_TRUE(writer.open().ok());
+ for (int group = 0; group < row_groups; ++group) {
+ auto column = type->create_column();
+ for (int row = 0; row < 100; ++row) {
+ column->insert(Field::create_field<TYPE_INT>(row % 3));
+ }
+ Block block;
+ block.insert({std::move(column), type, "value"});
+ ASSERT_TRUE(writer.write(block).ok());
+ }
+ ASSERT_TRUE(writer.close().ok());
+ std::vector<int64_t> offsets {999999};
+ ASSERT_TRUE(writer.collect_split_offsets_after_close(&offsets).ok());
+ ASSERT_EQ(row_groups, offsets.size());
+ auto reader = ::parquet::ParquetFileReader::OpenFile(_file_path,
false);
+ auto metadata = reader->metadata();
+ for (int i = 0; i < row_groups; ++i) {
+ auto row_group = metadata->RowGroup(i);
+ auto column = row_group->ColumnChunk(0);
+ ASSERT_EQ(dictionary, column->has_dictionary_page());
+ EXPECT_EQ(dictionary ? column->dictionary_page_offset() :
column->data_page_offset(),
+ offsets[i]);
+ EXPECT_GT(offsets[i], i == 0 ? 0 : offsets[i - 1]);
+ }
+ std::vector<int64_t> repeated;
+ ASSERT_TRUE(writer.collect_split_offsets_after_close(&repeated).ok());
+ EXPECT_EQ(offsets, repeated);
+ }
+
arrow::MemoryPool* _previous_pool = nullptr;
std::string _file_path;
std::shared_ptr<io::FileSystem> _fs;
};
+TEST_F(VParquetWriterTest, IcebergSplitOffsetsIncludeDictionaryPages) {
+ check_split_offsets(3, true);
+}
+
+TEST_F(VParquetWriterTest, IcebergSplitOffsetsWithoutDictionary) {
+ check_split_offsets(3, false);
+}
+
+TEST_F(VParquetWriterTest, IcebergSplitOffsetsSingleRowGroup) {
+ check_split_offsets(1, true);
+}
+
+TEST_F(VParquetWriterTest, IcebergSplitOffsetsEmptyFile) {
+ check_split_offsets(0, true);
+}
+
TEST_F(VParquetWriterTest, WritesIcebergPrimitiveAndCollectsMetrics) {
auto type = make_nullable(std::make_shared<DataTypeInt32>());
auto output_exprs = MockSlotRef::create_mock_contexts(DataTypes {type});
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
index e4709175800..fe1f1fd1da0 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
@@ -982,7 +982,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
return doPlanPositionDeletesSystemTableScan(handle, metadataTable,
columns, filter, session);
}
TableScan scan = buildScan(metadataTable, handle, filter, session);
- // Project the metadata-table scan to ONLY the requested columns, in
the SAME order BE lists them in
+ // Project the metadata-table scan to the requested columns first, in
the SAME order BE lists them in
// required_fields (== the scan slot order: buildColumnHandles
iterates desc.getSlots(), BE builds
// required_fields from the matching file_slot_descs). This mirrors
legacy IcebergScanNode
// .getSystemTableProjectedSchema() + scan.project(...). Two reasons
the projection is mandatory AND must
@@ -1011,6 +1011,13 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
projectedFields.add(field);
}
}
+ // Iceberg returns split_offsets only when the last offset is
smaller than file_size_in_bytes.
+ // Without that dependency, even valid offsets such as [4, 2048]
are hidden as null. Append it
+ // after the requested columns so BE's positional reads keep their
original order.
+ if (seenFieldIds.contains(DataFile.SPLIT_OFFSETS.fieldId())
+ && seenFieldIds.add(DataFile.FILE_SIZE.fieldId())) {
+
projectedFields.add(metadataSchema.findField(DataFile.FILE_SIZE.fieldId()));
+ }
scan = scan.project(new Schema(projectedFields));
}
List<ConnectorScanRange> ranges = new ArrayList<>();
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriterHelper.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriterHelper.java
index abf2edc38f2..06f5ceb4df0 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriterHelper.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriterHelper.java
@@ -184,7 +184,7 @@ final class IcebergWriterHelper {
partitionData =
Optional.of(convertToPartitionData(partitionValues, spec, zone));
}
DataFile dataFile = genDataFile(fileFormat, location, spec,
partitionData, recordCount, fileSize,
- metrics, sortOrder);
+ metrics, sortOrder, commitData.getSplitOffsets());
dataFiles.add(dataFile);
}
return WriteResult.builder()
@@ -194,7 +194,7 @@ final class IcebergWriterHelper {
private static DataFile genDataFile(FileFormat format, String location,
PartitionSpec spec,
Optional<PartitionData> partitionData, long recordCount, long
fileSize, Metrics metrics,
- SortOrder sortOrder) {
+ SortOrder sortOrder, List<Long> splitOffsets) {
DataFiles.Builder builder = DataFiles.builder(spec)
.withPath(location)
.withFileSizeInBytes(fileSize)
@@ -202,6 +202,10 @@ final class IcebergWriterHelper {
.withMetrics(metrics)
.withSortOrder(sortOrder)
.withFormat(format);
+ // Older BE fragments omit this optional field; only Parquet data
writers report these boundaries.
+ if (format == FileFormat.PARQUET && splitOffsets != null &&
!splitOffsets.isEmpty()) {
+ builder.withSplitOffsets(splitOffsets);
+ }
partitionData.ifPresent(builder::withPartition);
return builder.build();
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
index d997e826e05..bb46a63b355 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
@@ -3147,6 +3147,52 @@ public class IcebergScanPlanProviderTest {
+ "BufferUnderflows on complex/boolean bound columns
when materialised unrequested");
}
+ @Test
+ public void
planSystemTableScanKeepsSplitOffsetsWithTheirFileSizeDependency() throws
Exception {
+ List<Long> offsets = Arrays.asList(4L, 2048L);
+ for (PartitionSpec spec : Arrays.asList(PartitionSpec.unpartitioned(),
+ PartitionSpec.builderFor(SCHEMA).identity("id").build())) {
+ Table table = createTable("t1", SCHEMA, spec);
+ table.newAppend().appendFile(dataFile(table.spec(),
"data.parquet", 4096, offsets,
+ spec.isPartitioned() ? "id=1" : null)).commit();
+ IcebergScanPlanProvider provider = new IcebergScanPlanProvider(
+ IcebergCatalogProperties.of(Collections.emptyMap()),
opsReturning(table));
+ for (String type : Arrays.asList("files", "data_files",
"all_files", "all_data_files")) {
+ for (List<String> names :
Arrays.asList(Collections.singletonList("split_offsets"),
+ Arrays.asList("content", "split_offsets"),
Arrays.asList("split_offsets", "content"),
+ Arrays.asList("split_offsets", "file_size_in_bytes"),
+ Arrays.asList("file_size_in_bytes", "split_offsets")))
{
+ List<ConnectorColumnHandle> columns = names.stream()
+ .map(name -> (ConnectorColumnHandle) new
IcebergColumnHandle(name, 1))
+ .collect(Collectors.toList());
+ List<ConnectorScanRange> ranges = provider.planScan(null,
ConnectorScanRequest.builder(
+ IcebergTableHandle.forSystemTable("db1", "t1",
type, -1L, null, -1L), columns).build());
+ int rowCount = 0;
+ for (ConnectorScanRange range : ranges) {
+ FileScanTask task =
IcebergSystemTableSerialization.deserializeFromBase64(
+ ((IcebergScanRange)
range).getSerializedSplit());
+ try (CloseableIterable<StructLike> rows =
task.asDataTask().rows()) {
+ for (StructLike row : rows) {
+ Assertions.assertEquals(offsets,
row.get(names.indexOf("split_offsets"), List.class),
+ type + " " + names);
+ if (names.contains("content")) {
+ Assertions.assertEquals(0,
row.get(names.indexOf("content"), Integer.class));
+ }
+ if (names.contains("file_size_in_bytes")) {
+ Assertions.assertEquals(4096L,
+
row.get(names.indexOf("file_size_in_bytes"), Long.class));
+ }
+ rowCount++;
+ }
+ }
+
Assertions.assertNull(task.schema().findField("readable_metrics"));
+ }
+ Assertions.assertEquals(1, rowCount);
+ }
+ }
+ }
+ }
+
@Test
public void
planSystemTableScanProjectionPreservesRequestedColumnOrderForPositionalBeRead()
throws Exception {
// WHY (regression — External Regression build 1000283): BE's
IcebergSysTableJniScanner reads the
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriterHelperTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriterHelperTest.java
index 75160363940..c07f04896f5 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriterHelperTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWriterHelperTest.java
@@ -26,6 +26,7 @@ import org.apache.iceberg.DataFile;
import org.apache.iceberg.DataFiles;
import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -33,13 +34,17 @@ import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.WriteResult;
import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Types;
+import org.apache.thrift.TDeserializer;
+import org.apache.thrift.protocol.TJSONProtocol;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.Arrays;
@@ -212,6 +217,55 @@ public class IcebergWriterHelperTest {
Assertions.assertEquals(FileFormat.PARQUET, df.format());
}
+ @Test
+ public void convertToWriterResultCarriesSplitOffsetsFromCommitProtocol()
throws Exception {
+ Table table = tableWith("write.format.default", "parquet",
"write.metadata.metrics.default", "none");
+ TIcebergCommitData commit = new TIcebergCommitData();
+ // Decode the optional field on the wire so this also exercises
compatibility with older BE fragments.
+ String fragment =
"{\"1\":{\"str\":\"data.parquet\"},\"2\":{\"i64\":100},"
+ + "\"3\":{\"i64\":4096},\"15\":{\"lst\":[\"i64\",2,4,2048]}}";
+ new TDeserializer(new TJSONProtocol.Factory()).deserialize(commit,
fragment.getBytes(StandardCharsets.UTF_8));
+ DataFile file = IcebergWriterHelper.convertToWriterResult(
+ table, Collections.singletonList(commit),
ZoneOffset.UTC).dataFiles()[0];
+ Assertions.assertEquals(Arrays.asList(4L, 2048L), file.splitOffsets());
+ table.newAppend().appendFile(file).commit();
+ try (CloseableIterable<FileScanTask> tasks =
table.newScan().planFiles()) {
+ Assertions.assertEquals(Arrays.asList(4L, 2048L),
tasks.iterator().next().file().splitOffsets());
+ }
+ }
+
+ @Test
+ public void
convertToWriterResultKeepsOffsetsPerFileAndSupportsOldFragments() {
+ Table table = tableWith("write.format.default", "parquet");
+ List<TIcebergCommitData> commits = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ TIcebergCommitData commit = new TIcebergCommitData();
+ commit.setFilePath("data-" + i + ".parquet");
+ commit.setRowCount(10);
+ commit.setFileSize(4096);
+ if (i < 2) {
+ commit.setSplitOffsets(i == 0 ? Arrays.asList(4L, 1024L) :
Collections.singletonList(4L));
+ }
+ commits.add(commit);
+ }
+ DataFile[] files = IcebergWriterHelper.convertToWriterResult(table,
commits, ZoneOffset.UTC).dataFiles();
+ Assertions.assertEquals(Arrays.asList(4L, 1024L),
files[0].splitOffsets());
+ Assertions.assertEquals(Collections.singletonList(4L),
files[1].splitOffsets());
+ Assertions.assertNull(files[2].splitOffsets());
+ }
+
+ @Test
+ public void convertToWriterResultDoesNotApplyParquetOffsetsToOrc() {
+ TIcebergCommitData commit = new TIcebergCommitData();
+ commit.setFilePath("data.orc");
+ commit.setRowCount(10);
+ commit.setFileSize(4096);
+ commit.setSplitOffsets(Collections.singletonList(4L));
+ DataFile file =
IcebergWriterHelper.convertToWriterResult(tableWith("write.format.default",
"orc"),
+ Collections.singletonList(commit),
ZoneOffset.UTC).dataFiles()[0];
+ Assertions.assertNull(file.splitOffsets());
+ }
+
@Test
public void convertToWriterResultUsesPinnedMetadataAfterTableEvolution() {
// Files are labeled with the spec, sort order, format and metrics
policy the writer was pinned to, even after
diff --git a/gensrc/thrift/DataSinks.thrift b/gensrc/thrift/DataSinks.thrift
index 6ad86202e07..dd0dbabc51a 100644
--- a/gensrc/thrift/DataSinks.thrift
+++ b/gensrc/thrift/DataSinks.thrift
@@ -461,6 +461,8 @@ struct TIcebergCommitData {
13: optional i64 content_size_in_bytes
// Rows newly affected by the current statement for this commit item.
14: optional i64 affected_rows
+ // Parquet data-file row-group starts, independent of column metrics
policy.
+ 15: optional list<i64> split_offsets
}
struct TSortField {
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_partition_distribution.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_partition_distribution.groovy
index 6e355f33605..751f8caf352 100644
---
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_partition_distribution.groovy
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_partition_distribution.groovy
@@ -89,6 +89,11 @@ suite("test_iceberg_write_partition_distribution",
) partition_file_counts
"""
+ assertEquals(0L, (sql """
+ select count(*) from identity_partitions\$files
+ where content = 0 and (split_offsets is null or size(split_offsets) =
0 or split_offsets[1] <= 0)
+ """)[0][0] as long)
+
// Transform partitions: rows of one day carry different timestamps, so
routing by the raw source value would
// split a day across writers. The hash must cover day(ts).
sql """drop table if exists day_partitions"""
@@ -100,7 +105,8 @@ suite("test_iceberg_write_partition_distribution",
partition by list (day(ts)) ()
properties (
"format-version" = "2",
- "write.format.default" = "parquet"
+ "write.format.default" = "parquet",
+ "write.metadata.metrics.default" = "none"
)
"""
qt_day_plan """
@@ -125,4 +131,10 @@ suite("test_iceberg_write_partition_distribution",
group by `partition`
) partition_file_counts
"""
+ // Disabling column metrics must not suppress the physical split
boundaries in each data file.
+ assertEquals(0L, (sql """
+ select count(*) from day_partitions\$files
+ where content = 0 and (split_offsets is null or size(split_offsets) =
0 or split_offsets[1] <= 0)
+ """)[0][0] as long)
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
