From 035a843648df12bf814ff946c96f8d1eff09eb87 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Wed, 4 Nov 2026 13:06:31 +0800 Subject: [PATCH] [feature](be) Add session controls for Parquet page indexes (#69971) ### What problem does this PR solve? Parquet output relies on implicit page-index and data-page-size defaults, so users cannot explicitly omit indexes or tune the target page size. Add two validated session variables shared by OUTFILE, EXPORT, Hive and Iceberg Parquet writers: - `enable_parquet_write_page_index`: write ColumnIndex/OffsetIndex metadata; defaults to `true`, preserving the current master behavior. - `parquet_write_data_page_size`: target uncompressed data-page size in bytes; defaults to `1048576`, accepts positive 32-bit integers, and applies with indexes enabled or disabled. Keep the existing 128 KiB target cap for Hadoop-framed LZ4. Other codecs use the requested target. Row-group size, dictionary-page size, and data-page/format versions are unchanged. Optional query fields preserve existing defaults when omitted by older FEs. ### Release note Add session controls for Parquet page-index output and target data-page size. ### Validation - 16 BE ASAN tests passed, including real file metadata/readback, index on/off, dictionary on/off, null/all-null columns, Parquet versions, Data Page V1/V2, page offset bounds, Hadoop-framed LZ4 limits, and Hive/Iceberg partition writers. - 9 FE tests passed for defaults, variable forwarding, Thrift serialization, valid bounds, and invalid input. - The new tests failed for the missing settings before implementation. - FE Checkstyle, clang-format 16, build hygiene, and clang-tidy passed. - No cluster SQL run or performance benchmark was performed locally; CI validation is pending. ### Check List (For Author) - Test: - [x] Unit Test - Behavior changed: - [x] Yes. Explicit controls are available; current defaults are preserved. - Does this need documentation? - [x] Yes. English and Chinese documentation: https://github.com/apache/doris-website/pull/4581 ### Check List (For Reviewer who merge this PR) - [ ] Confirm the release note - [ ] Confirm test cases - [ ] Confirm document - [ ] Add branch pick label --- be/src/format/transformer/vparquet_writer.cpp | 19 +- .../transformer/vparquet_writer_test.cpp | 226 ++++++++++++++++++ .../org/apache/doris/qe/SessionVariable.java | 18 ++ .../qe/SessionVariableParquetWriterTest.java | 72 ++++++ gensrc/thrift/PaloInternalService.thrift | 3 + 5 files changed, 337 insertions(+), 1 deletion(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariableParquetWriterTest.java diff --git a/be/src/format/transformer/vparquet_writer.cpp b/be/src/format/transformer/vparquet_writer.cpp index c5a06867b15..d836a9a1b9c 100644 --- a/be/src/format/transformer/vparquet_writer.cpp +++ b/be/src/format/transformer/vparquet_writer.cpp @@ -27,6 +27,7 @@ #include #include +#include #include #include #include @@ -46,6 +47,9 @@ #include "util/timezone_utils.h" namespace doris { +namespace { +constexpr int64_t HADOOP_LZ4_DEFAULT_BUFFER_SIZE = 256 * 1024; +} // namespace ParquetOutputStream::ParquetOutputStream(doris::io::FileWriter* file_writer) : _file_writer(file_writer), _cur_pos(0), _written_len(0) { @@ -141,7 +145,6 @@ void ParquetBuildHelper::build_compression_type( break; } case TParquetCompressionType::LZ4_HADOOP: { - constexpr int64_t HADOOP_LZ4_DEFAULT_BUFFER_SIZE = 256 * 1024; // Hadoop-framed LZ4 -> Parquet thrift codec "LZ4" (deprecated). This matches what // Spark/Iceberg writes for `write.parquet.compression-codec=lz4`. Arrow 17 emits one // Hadoop LZ4 block per Parquet page/dictionary page, while Hadoop JVM readers default to @@ -213,6 +216,20 @@ Status VParquetWriter::_parse_properties() { ::parquet::WriterProperties::Builder builder; ParquetBuildHelper::build_compression_type(builder, _parquet_options.compression_type); ParquetBuildHelper::build_version(builder, _parquet_options.parquet_version); + const auto& query_options = _state->query_options(); + // Preserve the current enabled default explicitly, including when older FEs omit the field. + if (query_options.enable_parquet_write_page_index) { + builder.enable_write_page_index(); + } else { + builder.disable_write_page_index(); + } + int64_t page_size = query_options.parquet_write_data_page_size; + DORIS_CHECK_GT(page_size, 0); + // A 1 MiB session target must still respect the 128 KiB Hadoop-framed LZ4 limit. + if (_parquet_options.compression_type == TParquetCompressionType::LZ4_HADOOP) { + page_size = std::min(page_size, HADOOP_LZ4_DEFAULT_BUFFER_SIZE / 2); + } + builder.data_pagesize(page_size); if (_parquet_options.parquet_disable_dictionary) { builder.disable_dictionary(); } else { diff --git a/be/test/format/transformer/vparquet_writer_test.cpp b/be/test/format/transformer/vparquet_writer_test.cpp index 1c53c4b6916..85714050d23 100644 --- a/be/test/format/transformer/vparquet_writer_test.cpp +++ b/be/test/format/transformer/vparquet_writer_test.cpp @@ -22,6 +22,7 @@ #include #include #include +#include #include #include @@ -178,11 +179,236 @@ protected: EXPECT_EQ(offsets, repeated); } + Block page_index_block() { + auto type = std::make_shared(); + auto nullable = make_nullable(type); + auto ids = type->create_column(); + auto values = nullable->create_column(); + auto nulls = nullable->create_column(); + for (int32_t row = 0; row < 12000; ++row) { + ids->insert(Field::create_field(row)); + if (row % 7 == 0) { + values->insert_default(); + } else { + values->insert(Field::create_field(row % 64)); + } + nulls->insert_default(); + } + return Block({{std::move(ids), type, "id"}, + {std::move(values), nullable, "value"}, + {std::move(nulls), nullable, "all_null"}}); + } + + void check_page_indexes(bool enabled) { + auto reader = ::parquet::ParquetFileReader::OpenFile(_file_path, false); + ASSERT_EQ(reader->metadata()->num_row_groups(), 1); + auto group = reader->metadata()->RowGroup(0); + ASSERT_EQ(group->num_rows(), 12000); + for (int column = 0; column < group->num_columns(); ++column) { + auto chunk = group->ColumnChunk(column); + EXPECT_EQ(chunk->GetColumnIndexLocation().has_value(), enabled); + EXPECT_EQ(chunk->GetOffsetIndexLocation().has_value(), enabled); + if (!enabled) { + continue; + } + auto index_reader = reader->GetPageIndexReader(); + ASSERT_NE(index_reader, nullptr); + auto indexes = index_reader->RowGroup(0); + ASSERT_NE(indexes, nullptr); + auto column_index = indexes->GetColumnIndex(column); + auto offset_index = indexes->GetOffsetIndex(column); + ASSERT_NE(column_index, nullptr); + ASSERT_NE(offset_index, nullptr); + const auto& pages = offset_index->page_locations(); + ASSERT_FALSE(pages.empty()); + EXPECT_EQ(pages.size(), column_index->null_pages().size()); + EXPECT_EQ(pages.front().offset, chunk->data_page_offset()); + EXPECT_EQ(pages.front().first_row_index, 0); + const auto start = chunk->has_dictionary_page() ? chunk->dictionary_page_offset() + : chunk->data_page_offset(); + int64_t previous_row = -1; + int64_t previous_end = start; + for (const auto& page : pages) { + EXPECT_GT(page.first_row_index, previous_row); + EXPECT_LT(page.first_row_index, group->num_rows()); + EXPECT_GE(page.offset, previous_end); + EXPECT_GT(page.compressed_page_size, 0); + EXPECT_LE(page.offset + page.compressed_page_size, + start + chunk->total_compressed_size()); + previous_row = page.first_row_index; + previous_end = page.offset + page.compressed_page_size; + } + if (column == 0) { + EXPECT_GT(pages.size(), 1); + } else if (column == 2) { + EXPECT_TRUE(std::all_of(column_index->null_pages().begin(), + column_index->null_pages().end(), + [](bool value) { return value; })); + } + } + auto input = arrow::io::ReadableFile::Open(_file_path); + ASSERT_TRUE(input.ok()) << input.status(); + auto arrow_reader = ::parquet::arrow::OpenFile(*input, arrow::default_memory_pool()); + ASSERT_TRUE(arrow_reader.ok()) << arrow_reader.status(); + auto table = (*arrow_reader)->ReadTable(); + ASSERT_TRUE(table.ok()) << table.status(); + EXPECT_EQ((*table)->num_rows(), 12000); + EXPECT_EQ((*table)->column(1)->null_count(), (12000 + 6) / 7); + EXPECT_EQ((*table)->column(2)->null_count(), 12000); + const auto& ids = assert_cast(*(*table)->column(0)->chunk(0)); + EXPECT_EQ(ids.Value(0), 0); + EXPECT_EQ(ids.Value(11999), 11999); + } + arrow::MemoryPool* _previous_pool = nullptr; std::string _file_path; std::shared_ptr _fs; }; +TEST_F(VParquetWriterTest, ParquetPageIndexToggle) { + for (const bool enabled : {false, true}) { + for (const bool dictionary : {false, true}) { + for (const auto version : + {TParquetVersion::PARQUET_1_0, TParquetVersion::PARQUET_2_LATEST}) { + TQueryOptions query_options; + query_options.__set_enable_parquet_write_page_index(enabled); + query_options.__set_parquet_write_data_page_size(1024); + RuntimeState state(query_options, TQueryGlobals()); + state.set_timezone("UTC"); + auto block = page_index_block(); + auto exprs = MockSlotRef::create_mock_contexts(block.get_data_types()); + io::FileWriterPtr file; + ASSERT_TRUE(_fs->create_file(_file_path, &file).ok()); + ParquetFileOptions options {.compression_type = TParquetCompressionType::SNAPPY, + .parquet_version = version, + .parquet_disable_dictionary = !dictionary}; + VParquetWriter writer(&state, file.get(), exprs, + std::vector {"id", "value", "all_null"}, false, + options); + ASSERT_TRUE(writer.open().ok()); + ASSERT_TRUE(writer.write(block).ok()); + ASSERT_TRUE(writer.close().ok()); + check_page_indexes(enabled); + } + } + } +} + +TEST_F(VParquetWriterTest, ParquetPageIndexDataPageVersions) { + for (const auto page_version : + {::parquet::ParquetDataPageVersion::V1, ::parquet::ParquetDataPageVersion::V2}) { + TQueryOptions query_options; + query_options.__set_enable_parquet_write_page_index(true); + query_options.__set_parquet_write_data_page_size(1024); + RuntimeState state(query_options, TQueryGlobals()); + auto block = page_index_block(); + auto exprs = MockSlotRef::create_mock_contexts(block.get_data_types()); + io::FileWriterPtr file; + ASSERT_TRUE(_fs->create_file(_file_path, &file).ok()); + const std::vector names {"id", "value", "all_null"}; + ParquetFileOptions options {.compression_type = TParquetCompressionType::SNAPPY, + .parquet_version = TParquetVersion::PARQUET_2_LATEST}; + VParquetWriter writer(&state, file.get(), exprs, names, false, options); + ASSERT_TRUE(writer._parse_properties().ok()); + // Exercise both page layouts with the production writer properties; this does not add + // a SQL option for changing the existing data-page version. + auto properties = ::parquet::WriterProperties::Builder(*writer._parquet_writer_properties) + .data_page_version(page_version) + ->build(); + auto converter = writer._create_arrow_block_convertor(block.get_data_types(), names, "UTC", + cctz::utc_time_zone(), false); + ASSERT_TRUE(converter->init().ok()); + std::shared_ptr batch; + ASSERT_TRUE(converter->convert_to_arrow(block, arrow::default_memory_pool(), &batch).ok()); + auto result = ::parquet::arrow::FileWriter::Open( + *batch->schema(), arrow::default_memory_pool(), writer._outstream, properties, + writer._arrow_properties); + ASSERT_TRUE(result.ok()) << result.status(); + auto arrow_writer = std::move(result).ValueOrDie(); + ASSERT_TRUE(arrow_writer->WriteRecordBatch(*batch).ok()); + ASSERT_TRUE(arrow_writer->Close().ok()); + ASSERT_TRUE(writer._outstream->Close().ok()); + check_page_indexes(true); + } +} + +TEST_F(VParquetWriterTest, ParquetPageSizePreservesHadoopLz4Limit) { + for (const auto codec : {TParquetCompressionType::LZ4_HADOOP, TParquetCompressionType::LZ4}) { + for (const int32_t requested_size : {65536, 262144}) { + TQueryOptions query_options; + query_options.__set_parquet_write_data_page_size(requested_size); + RuntimeState state(query_options, TQueryGlobals()); + state.set_timezone("UTC"); + auto block = page_index_block(); + auto exprs = MockSlotRef::create_mock_contexts(block.get_data_types()); + io::FileWriterPtr file; + ASSERT_TRUE(_fs->create_file(_file_path, &file).ok()); + ParquetFileOptions options {.compression_type = codec, + .parquet_version = TParquetVersion::PARQUET_1_0}; + VParquetWriter writer(&state, file.get(), exprs, + std::vector {"id", "value", "all_null"}, false, + options); + ASSERT_TRUE(writer.open().ok()); + EXPECT_TRUE(writer._parquet_writer_properties->page_index_enabled()); + EXPECT_EQ(writer._parquet_writer_properties->data_pagesize(), + codec == TParquetCompressionType::LZ4_HADOOP + ? std::min(requested_size, 131072) + : requested_size); + ASSERT_TRUE(writer.close().ok()); + } + } +} + +TEST_F(VParquetWriterTest, HiveAndIcebergParquetPageIndexOptions) { + const std::string json = + R"({"type":"struct","fields":[{"id":1,"name":"id","required":true,"type":"int"},{"id":2,"name":"value","required":false,"type":"int"},{"id":3,"name":"all_null","required":false,"type":"int"}]})"; + auto schema = iceberg::SchemaParser::from_json(json); + const std::map hadoop_conf; + for (const bool iceberg_writer : {false, true}) { + for (const bool enabled : {false, true}) { + TQueryOptions query_options; + query_options.__set_enable_parquet_write_page_index(enabled); + query_options.__set_parquet_write_data_page_size(1024); + RuntimeState state(query_options, TQueryGlobals()); + state.set_timezone("UTC"); + auto block = page_index_block(); + auto exprs = MockSlotRef::create_mock_contexts(block.get_data_types()); + const auto name = "page_index_" + UniqueId::gen_uid().to_string(); + _file_path = "./" + name + "-0.parquet"; + TDataSink sink; + if (iceberg_writer) { + VIcebergPartitionWriter writer(sink, {}, exprs, *schema, &json, + {"id", "value", "all_null"}, + {.write_path = ".", + .original_write_path = ".", + .target_path = ".", + .file_type = TFileType::FILE_LOCAL, + .broker_addresses = {}}, + name, 0, TFileFormatType::FORMAT_PARQUET, + TFileCompressType::PLAIN, hadoop_conf); + ASSERT_TRUE(writer.open(&state, nullptr, nullptr).ok()); + ASSERT_TRUE(writer.write(block).ok()); + ASSERT_TRUE(writer.close(Status::OK()).ok()); + } else { + VHivePartitionWriter writer(sink, "", TUpdateMode::APPEND, exprs, + {"id", "value", "all_null"}, + {.write_path = ".", + .original_write_path = ".", + .target_path = ".", + .file_type = TFileType::FILE_LOCAL, + .broker_addresses = {}}, + name, 0, TFileFormatType::FORMAT_PARQUET, + TFileCompressType::PLAIN, nullptr, hadoop_conf); + ASSERT_TRUE(writer.open(&state, nullptr).ok()); + ASSERT_TRUE(writer.write(block).ok()); + ASSERT_TRUE(writer.close(Status::OK()).ok()); + } + check_page_indexes(enabled); + ASSERT_TRUE(_fs->delete_file(_file_path).ok()); + } + } +} + TEST_F(VParquetWriterTest, IcebergSplitOffsetsIncludeDictionaryPages) { check_split_offsets(3, true); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index 15a28aaa33a..68129651c4c 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java @@ -564,6 +564,8 @@ public class SessionVariable implements Serializable, Writable { public static final String ENABLE_PAGE_CACHE = "enable_page_cache"; public static final String ENABLE_PARQUET_FILE_PAGE_CACHE = "enable_parquet_file_page_cache"; + public static final String ENABLE_PARQUET_WRITE_PAGE_INDEX = "enable_parquet_write_page_index"; + public static final String PARQUET_WRITE_DATA_PAGE_SIZE = "parquet_write_data_page_size"; public static final String DUMP_NEREIDS_MEMO = "dump_nereids_memo"; @@ -2568,6 +2570,19 @@ public class SessionVariable implements Serializable, Writable { needForward = true) public boolean enableParquetFilePageCache = true; + @VarAttrDef.VarAttr(name = ENABLE_PARQUET_WRITE_PAGE_INDEX, needForward = true, + description = "Write ColumnIndex and OffsetIndex metadata to new Parquet files. Default is true.") + public boolean enableParquetWritePageIndex = true; + + @VarAttrDef.VarAttr(name = PARQUET_WRITE_DATA_PAGE_SIZE, needForward = true, + checker = "checkParquetWriteDataPageSize", + description = "Target uncompressed Parquet data-page size in bytes. Must be positive; default is 1048576.") + public int parquetWriteDataPageSize = 1048576; + + public void checkParquetWriteDataPageSize(String value) throws Exception { + checkFieldValue(PARQUET_WRITE_DATA_PAGE_SIZE, 1, value); + } + @VarAttrDef.VarAttr(name = ENABLE_FOLD_NONDETERMINISTIC_FN) public boolean enableFoldNondeterministicFn = false; @@ -5845,6 +5860,9 @@ public class SessionVariable implements Serializable, Writable { } tResult.setEnableParquetLazyMat(enableParquetLazyMat); + // Query options carry the same write policy to OUTFILE and catalog sinks. + tResult.setEnableParquetWritePageIndex(enableParquetWritePageIndex); + tResult.setParquetWriteDataPageSize(parquetWriteDataPageSize); tResult.setEnableOrcLazyMat(enableOrcLazyMat); tResult.setEnableParquetFilterByMinMax(enableParquetFilterByMinMax); tResult.setEnableParquetFilterByBloomFilter(enableParquetFilterByBloomFilter); diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariableParquetWriterTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariableParquetWriterTest.java new file mode 100644 index 00000000000..f1bcf597751 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariableParquetWriterTest.java @@ -0,0 +1,72 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.qe; + +import org.apache.doris.analysis.SetType; +import org.apache.doris.analysis.SetVar; +import org.apache.doris.analysis.StringLiteral; +import org.apache.doris.common.DdlException; +import org.apache.doris.thrift.TQueryOptions; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +class SessionVariableParquetWriterTest { + private static final String WRITE_INDEX = "enable_parquet_write_page_index"; + private static final String PAGE_SIZE = "parquet_write_data_page_size"; + + @Test + void testDefaults() { + TQueryOptions options = new SessionVariable().toThrift(); + Assertions.assertTrue(options.isEnableParquetWritePageIndex()); + Assertions.assertEquals(1048576, options.getParquetWriteDataPageSize()); + } + + @Test + void testForwardingAndThriftSerialization() throws Exception { + SessionVariable session = new SessionVariable(); + VariableMgr.setVar(session, new SetVar(SetType.SESSION, WRITE_INDEX, new StringLiteral("false"))); + VariableMgr.setVar(session, new SetVar(SetType.SESSION, PAGE_SIZE, new StringLiteral("262144"))); + SessionVariable forwarded = new SessionVariable(); + forwarded.setForwardedSessionVariables(session.getForwardVariables()); + TQueryOptions options = forwarded.toThrift(); + Assertions.assertTrue(options.isSetEnableParquetWritePageIndex()); + Assertions.assertFalse(options.isEnableParquetWritePageIndex()); + Assertions.assertTrue(options.isSetParquetWriteDataPageSize()); + Assertions.assertEquals(262144, options.getParquetWriteDataPageSize()); + } + + @ParameterizedTest + @ValueSource(strings = {"1", "1048576", "2147483647"}) + void testValidPageSize(String value) throws Exception { + SessionVariable session = new SessionVariable(); + VariableMgr.setVar(session, new SetVar(SetType.SESSION, PAGE_SIZE, new StringLiteral(value))); + Assertions.assertEquals(Integer.parseInt(value), session.toThrift().getParquetWriteDataPageSize()); + } + + @ParameterizedTest + @ValueSource(strings = {"0", "-1", "2147483648", "invalid"}) + void testInvalidPageSize(String value) { + SessionVariable session = new SessionVariable(); + Assertions.assertThrows(DdlException.class, () -> VariableMgr.setVar(session, + new SetVar(SetType.SESSION, PAGE_SIZE, new StringLiteral(value)))); + Assertions.assertEquals(1048576, session.toThrift().getParquetWriteDataPageSize()); + } +} diff --git a/gensrc/thrift/PaloInternalService.thrift b/gensrc/thrift/PaloInternalService.thrift index ee15ddaa4c7..108faf3f94b 100644 --- a/gensrc/thrift/PaloInternalService.thrift +++ b/gensrc/thrift/PaloInternalService.thrift @@ -537,6 +537,9 @@ struct TQueryOptions { 236: optional bool supports_external_file_report_ack = false; // Memory limit for one non-streaming analytic partition before it proactively spills. 237: optional i64 spill_analytic_sink_mem_limit_bytes = 67108864; + // Optional Parquet output indexes; absent fields retain the existing writer defaults. + 238: optional bool enable_parquet_write_page_index = true; + 239: optional i32 parquet_write_data_page_size = 1048576; // For cloud, to control if the content would be written into file cache // In write path, to control if the content would be written into file cache. // In read path, read from file cache or remote storage when execute query.