feat: Migrate lookup file support for pk compact

src/paimon/core/mergetree/lookup/default_lookup_serializer_factory.h

@@ -0,0 +1,67 @@
+/*
+ * 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.
+ */
+
+#pragma once
+#include "paimon/common/data/serializer/row_compacted_serializer.h"
+#include "paimon/common/utils/arrow/arrow_utils.h"
+#include "paimon/core/mergetree/lookup/lookup_serializer_factory.h"
+namespace paimon {
+/// A `LookupSerializerFactory` using `RowCompactedSerializer`.
+class DefaultLookupSerializerFactory : public LookupSerializerFactory {
+ public:
+    static constexpr char kVersion[] = "v1";
+
+    std::string Version() const override {
+        return kVersion;
+    }
+
+    Result<SerializeFunc> CreateSerializer(const std::shared_ptr<arrow::Schema>& schema,
+                                           const std::shared_ptr<MemoryPool>& pool) const override {
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<RowCompactedSerializer> serializer,
+                               RowCompactedSerializer::Create(schema, pool));
+        return LookupSerializerFactory::SerializeFunc(
+            [serializer =
+                 std::move(serializer)](const InternalRow& row) -> Result<std::shared_ptr<Bytes>> {
+                return serializer->SerializeToBytes(row);
+            });
+    }
+
+    Result<DeserializeFunc> CreateDeserializer(
+        const std::string& file_ser_version, const std::shared_ptr<arrow::Schema>& current_schema,
+        const std::shared_ptr<arrow::Schema>& file_schema,
+        const std::shared_ptr<MemoryPool>& pool) const override {
+        if (Version() != file_ser_version) {
+            return Status::Invalid(fmt::format(
+                "file_ser_version {} mismatch DefaultLookupSerializerFactory version {}",
+                file_ser_version, Version()));
+        }
+        if (!ArrowUtils::EqualsIgnoreNullable(arrow::struct_(file_schema->fields()),
+                                              arrow::struct_(current_schema->fields()))) {
+            return Status::Invalid(
+                fmt::format("current_schema {} must be equal with file_schema {} in "
+                            "DefaultLookupSerializerFactory",
+                            current_schema->ToString(), file_schema->ToString()));
+        }
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<RowCompactedSerializer> serializer,
+                               RowCompactedSerializer::Create(current_schema, pool));
+        return LookupSerializerFactory::DeserializeFunc(
+            [serializer = std::move(serializer)](const std::shared_ptr<Bytes>& bytes)
+                -> Result<std::unique_ptr<InternalRow>> { return serializer->Deserialize(bytes); });
+    }
+};
+}  // namespace paimon

src/paimon/core/mergetree/lookup/file_position.h

@@ -0,0 +1,28 @@
+/*
+ * 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.
+ */
+
+#pragma once
+#include <cstdint>
+#include <string>
+namespace paimon {
+/// File name and row position for DeletionVector.
+struct FilePosition {
+    std::string file_name;
+    int64_t row_position;
+};
+}  // namespace paimon

src/paimon/core/mergetree/lookup/lookup_serializer_factory.h

@@ -0,0 +1,48 @@
+/*
+ * 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.
+ */
+
+#pragma once
+#include <functional>
+
+#include "arrow/api.h"
+#include "paimon/common/data/internal_row.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/result.h"
+namespace paimon {
+/// Factory to create serializer for lookup.
+class LookupSerializerFactory {
+ public:
+    virtual ~LookupSerializerFactory() = default;
+
+    virtual std::string Version() const = 0;
+
+    using SerializeFunc = std::function<Result<std::shared_ptr<Bytes>>(const InternalRow&)>;
+    using DeserializeFunc =
+        std::function<Result<std::unique_ptr<InternalRow>>(const std::shared_ptr<Bytes>&)>;
+
+    virtual Result<SerializeFunc> CreateSerializer(
+        const std::shared_ptr<arrow::Schema>& schema,
+        const std::shared_ptr<MemoryPool>& pool) const = 0;
+
+    virtual Result<DeserializeFunc> CreateDeserializer(
+        const std::string& file_ser_version, const std::shared_ptr<arrow::Schema>& current_schema,
+        const std::shared_ptr<arrow::Schema>& file_schema,
+        const std::shared_ptr<MemoryPool>& pool) const = 0;
+};
+}  // namespace paimon

src/paimon/core/mergetree/lookup/persist_empty_processor.h

@@ -0,0 +1,59 @@
+/*
+ * 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.
+ */
+
+#pragma once
+#include "paimon/core/mergetree/lookup/lookup_serializer_factory.h"
+#include "paimon/core/mergetree/lookup/persist_processor.h"
+namespace paimon {
+/// A `PersistProcessor` to return `bool` only.
+class PersistEmptyProcessor : public PersistProcessor<bool> {
+ public:
+    bool WithPosition() const override {
+        return false;
+    }
+
+    Result<std::shared_ptr<Bytes>> PersistToDisk(const KeyValue& kv) const override {
+        return Bytes::EmptyBytes();
+    }
+
+    Result<bool> ReadFromDisk(std::shared_ptr<InternalRow> key, int32_t level,
+                              const std::shared_ptr<Bytes>& value_bytes,
+                              const std::string& file_name) const override {
+        return true;
+    }
+
+    /// Factory to create `PersistProcessor`.
+    class Factory : public PersistProcessor<bool>::Factory {
+     public:
+        std::string Identifier() const override {
+            return "empty";
+        }
+
+        Result<std::unique_ptr<PersistProcessor<bool>>> Create(
+            const std::string& file_ser_version,
+            const std::shared_ptr<LookupSerializerFactory>& serializer_factory,
+            const std::shared_ptr<arrow::Schema>& file_schema,
+            const std::shared_ptr<MemoryPool>& pool) const override {
+            return std::unique_ptr<PersistEmptyProcessor>(new PersistEmptyProcessor());
+        }
+    };
+
+ private:
+    PersistEmptyProcessor() = default;
+};
+}  // namespace paimon

src/paimon/core/mergetree/lookup/persist_position_processor.h

@@ -0,0 +1,79 @@
+/*
+ * 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.
+ */
+
+#pragma once
+#include "paimon/common/utils/var_length_int_utils.h"
+#include "paimon/core/mergetree/lookup/file_position.h"
+#include "paimon/core/mergetree/lookup/lookup_serializer_factory.h"
+#include "paimon/core/mergetree/lookup/persist_processor.h"
+
+namespace paimon {
+/// A `PersistProcessor` to return `FilePosition`.
+class PersistPositionProcessor : public PersistProcessor<FilePosition> {
+ public:
+    bool WithPosition() const override {
+        return true;
+    }
+
+    Result<std::shared_ptr<Bytes>> PersistToDisk(const KeyValue& kv) const override {
+        return Status::Invalid(
+            "invalid operation, do not support persist to disk without position in "
+            "PersistPositionProcessor");
+    }
+
+    Result<std::shared_ptr<Bytes>> PersistToDisk(const KeyValue& kv,
+                                                 int64_t row_position) const override {
+        auto bytes = std::make_shared<Bytes>(VarLengthIntUtils::kMaxVarLongSize, pool_.get());
+        PAIMON_ASSIGN_OR_RAISE(int64_t len,
+                               VarLengthIntUtils::EncodeLong(row_position, bytes->data()));
+        std::shared_ptr<Bytes> copy_bytes = Bytes::CopyOf(*bytes, len, pool_.get());
+        return copy_bytes;
+    }
+
+    Result<FilePosition> ReadFromDisk(std::shared_ptr<InternalRow> key, int32_t level,
+                                      const std::shared_ptr<Bytes>& value_bytes,
+                                      const std::string& file_name) const override {
+        int32_t decode_offset = 0;
+        PAIMON_ASSIGN_OR_RAISE(int64_t row_position,
+                               VarLengthIntUtils::DecodeLong(value_bytes->data(), &decode_offset));
+        return FilePosition{file_name, row_position};
+    }
+
+    /// Factory to create `PersistProcessor`.
+    class Factory : public PersistProcessor<FilePosition>::Factory {
+     public:
+        std::string Identifier() const override {
+            return "position";
+        }
+
+        Result<std::unique_ptr<PersistProcessor<FilePosition>>> Create(
+            const std::string& file_ser_version,
+            const std::shared_ptr<LookupSerializerFactory>& serializer_factory,
+            const std::shared_ptr<arrow::Schema>& file_schema,
+            const std::shared_ptr<MemoryPool>& pool) const override {
+            return std::unique_ptr<PersistPositionProcessor>(new PersistPositionProcessor(pool));
+        }
+    };
+
+ private:
+    explicit PersistPositionProcessor(const std::shared_ptr<MemoryPool>& pool) : pool_(pool) {}
+
+ private:
+    std::shared_ptr<MemoryPool> pool_;
+};
+}  // namespace paimon

src/paimon/core/mergetree/lookup/persist_processor.h

@@ -0,0 +1,56 @@
+/*
+ * 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.
+ */
+
+#pragma once
+#include "paimon/core/key_value.h"
+#include "paimon/memory/bytes.h"
+namespace paimon {
+class LookupSerializerFactory;
+/// Processor to process value.
+template <typename T>
+class PersistProcessor {
+ public:
+    virtual ~PersistProcessor() = default;
+
+    virtual bool WithPosition() const = 0;
+
+    virtual Result<std::shared_ptr<Bytes>> PersistToDisk(const KeyValue& kv) const = 0;
+
+    virtual Result<std::shared_ptr<Bytes>> PersistToDisk(const KeyValue& kv,
+                                                         int64_t row_position) const {
+        return Status::Invalid("Not support for PersistToDisk with position");
+    }
+
+    virtual Result<T> ReadFromDisk(std::shared_ptr<InternalRow> key, int32_t level,
+                                   const std::shared_ptr<Bytes>& value_bytes,
+                                   const std::string& file_name) const = 0;
+
+    /// Factory to create `PersistProcessor`.
+    class Factory {
+     public:
+        virtual ~Factory() = default;
+        virtual std::string Identifier() const = 0;
+
+        virtual Result<std::unique_ptr<PersistProcessor<T>>> Create(
+            const std::string& file_ser_version,
+            const std::shared_ptr<LookupSerializerFactory>& serializer_factory,
+            const std::shared_ptr<arrow::Schema>& file_schema,
+            const std::shared_ptr<MemoryPool>& pool) const = 0;
+    };
+};
+}  // namespace paimon