[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

@@ -27,6 +27,7 @@
 #include <parquet/type_fwd.h>
 #include <parquet/types.h>
 
+#include <algorithm>
 #include <ctime>
 #include <exception>
 #include <ostream>
@@ -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 {

be/test/format/transformer/vparquet_writer_test.cpp

@@ -22,6 +22,7 @@
 #include <gtest/gtest.h>
 #include <parquet/api/reader.h>
 #include <parquet/arrow/reader.h>
+#include <parquet/page_index.h>
 #include <parquet/schema.h>
 
 #include <algorithm>
@@ -178,11 +179,236 @@ protected:
         EXPECT_EQ(offsets, repeated);
     }
 
+    Block page_index_block() {
+        auto type = std::make_shared<DataTypeInt32>();
+        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<TYPE_INT>(row));
+            if (row % 7 == 0) {
+                values->insert_default();
+            } else {
+                values->insert(Field::create_field<TYPE_INT>(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<const arrow::Int32Array&>(*(*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<io::FileSystem> _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<std::string> {"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<std::string> 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<arrow::RecordBatch> 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<std::string> {"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<std::string, std::string> 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);
 }

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);

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());
+    }
+}

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.