fix review comments

Author: zhangweilst

src/paimon/common/predicate/literal_converter.h

@@ -134,10 +134,6 @@ class PAIMON_EXPORT LiteralConverter {
                 literals.emplace_back(literal_type);
             } else {
                 int64_t dict_index = indices->Value(i);
-                if (dictionary->IsNull(dict_index)) {
-                    literals.emplace_back(literal_type);
-                    continue;
-                }
                 auto value = dictionary->GetView(dict_index);
                 literals.emplace_back(literal_type, value.data(), value.size(), own_data);
             }

src/paimon/core/casting/casting_utils.cpp

@@ -30,6 +30,8 @@ Result<std::shared_ptr<arrow::Array>> CastingUtils::DecodeDictionary(
     }
     const auto& dictionary_type = checked_cast<const arrow::DictionaryType&>(*array->type());
     std::shared_ptr<arrow::DataType> value_type = dictionary_type.value_type();
+    /// We're not casting `LARGE_BINARY` to `BINARY` since `BLOB` specifically expects
+    /// `LARGE_BINARY`. See also in `src/paimon/common/utils/field_type_utils.h`.
     if (value_type->id() == arrow::Type::LARGE_STRING) {
         value_type = arrow::utf8();
     }

src/paimon/format/parquet/parquet_read_type_adapter.cpp

@@ -25,6 +25,7 @@
 #include "arrow/type.h"
 #include "fmt/format.h"
 #include "paimon/common/data/blob_utils.h"
+#include "paimon/common/utils/arrow/arrow_utils.h"
 #include "paimon/common/utils/arrow/status_utils.h"
 #include "paimon/common/utils/checked_cast.h"
 #include "paimon/common/utils/date_time_utils.h"
@@ -100,6 +101,17 @@ Result<bool> ParquetReadTypeAdapter::NeedsArrayConversionImpl(
     const std::shared_ptr<arrow::DataType>& src_data_type = src_field->type();
     const std::shared_ptr<arrow::DataType>& target_data_type = target_field->type();
     arrow::Type::type type = src_data_type->id();
+    if (type == arrow::Type::DICTIONARY) {
+        // ARROW:schema may restore a Parquet leaf as a dictionary even though the Paimon schema
+        // describes its logical value type. Keep the encoded representation when its value
+        // layout is supported and already matches that logical type.
+        const auto& dictionary_type = checked_cast<const arrow::DictionaryType&>(*src_data_type);
+        const std::shared_ptr<arrow::DataType>& value_type = dictionary_type.value_type();
+        if (ArrowUtils::IsDictionaryLayoutRecoverableValueType(*value_type) &&
+            value_type->Equals(target_data_type)) {
+            return false;
+        }
+    }
     if (type != target_data_type->id()) {
         return Status::Invalid(fmt::format("source type {} and target type {} mismatch",
                                            src_data_type->ToString(),

src/paimon/format/parquet/parquet_read_type_adapter_test.cpp

@@ -109,6 +109,24 @@ TEST(ParquetReadTypeAdapterTest, TestAlreadyLogicalBlobCompatibility) {
                         "source type large_binary and target type binary mismatch");
 }
 
+TEST(ParquetReadTypeAdapterTest, TestDictionaryBinaryCompatibility) {
+    std::shared_ptr<arrow::DataType> dictionary_type =
+        arrow::dictionary(arrow::int32(), arrow::binary());
+
+    ASSERT_OK_AND_ASSIGN(bool needs_conversion, ParquetReadTypeAdapter::NeedsArrayConversion(
+                                                    dictionary_type, arrow::binary()));
+    ASSERT_FALSE(needs_conversion);
+
+    ASSERT_NOK_WITH_MSG(
+        ParquetReadTypeAdapter::NeedsArrayConversion(dictionary_type, arrow::utf8()),
+        "source type dictionary<values=binary, indices=int32, ordered=0> and target type string "
+        "mismatch");
+    ASSERT_NOK_WITH_MSG(ParquetReadTypeAdapter::NeedsArrayConversion(
+                            arrow::dictionary(arrow::int32(), arrow::int32()), arrow::int32()),
+                        "source type dictionary<values=int32, indices=int32, ordered=0> and "
+                        "target type int32 mismatch");
+}
+
 TEST(ParquetReadTypeAdapterTest, TestAdaptArray) {
     auto timezone = DateTimeUtils::GetLocalTimezoneName();
     arrow::FieldVector fields = {

test/inte/scan_and_read_inte_test.cpp

@@ -3424,6 +3424,41 @@ TEST_P(ScanAndReadInteTest, TestCastTimestampType) {
     ASSERT_TRUE(expected->Equals(read_result)) << read_result->ToString();
 }
 
+TEST_F(ScanAndReadInteTest, TestParquetDictionaryBinaryStoredArrowSchema) {
+    const std::string table_path = GetDataDir() +
+                                   "/parquet/append_dictionary_binary.db/"
+                                   "append_dictionary_binary";
+
+    ScanContextBuilder scan_context_builder(table_path);
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<ScanContext> scan_context, scan_context_builder.Finish());
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableScan> table_scan,
+                         TableScan::Create(std::move(scan_context)));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> plan, table_scan->CreatePlan());
+    ASSERT_EQ(1, plan->SnapshotId().value());
+    ASSERT_EQ(1, plan->Splits().size());
+
+    ReadContextBuilder read_context_builder(table_path);
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> read_context, read_context_builder.Finish());
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> table_read,
+                         TableRead::Create(std::move(read_context)));
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> reader,
+                         table_read->CreateReader(plan->Splits()));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> actual,
+                         ReadResultCollector::CollectResult(std::move(reader)));
+
+    std::shared_ptr<arrow::DataType> expected_type = arrow::struct_({
+        arrow::field("_VALUE_KIND", arrow::int8()),
+        arrow::field("payload", arrow::binary()),
+    });
+    std::shared_ptr<arrow::Array> expected_array =
+        arrow::ipc::internal::json::ArrayFromJSON(
+            expected_type, R"([[0, "alpha"], [0, null], [0, "beta"], [0, "alpha"]])")
+            .ValueOrDie();
+    std::shared_ptr<arrow::ChunkedArray> expected =
+        std::make_shared<arrow::ChunkedArray>(expected_array);
+    ASSERT_TRUE(expected->Equals(actual)) << actual->ToString();
+}
+
 #ifdef PAIMON_ENABLE_LANCE
 TEST_F(ScanAndReadInteTest, TestLanceJavaCompatibility) {
     TimezoneGuard timezone_guard("UTC");

test/test_data/parquet/append_dictionary_binary.db/append_dictionary_binary/README

@@ -0,0 +1,12 @@
+Paimon schema:
+payload: BYTES
+
+The Parquet file was written from an Arrow dictionary<int32, binary> array with
+ArrowWriterProperties::store_schema enabled. Its serialized ARROW:schema metadata therefore
+restores payload as dictionary<int32, binary>, while the Paimon table schema remains BYTES.
+
+Rows:
+alpha
+null
+beta
+alpha