hudi-agent commented on code in PR #19765: URL: https://github.com/apache/hudi/pull/19765#discussion_r3869195302
########## hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/PartitionedIndexBackendFactory.java: ########## @@ -0,0 +1,54 @@ +/* + * 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.hudi.sink.partitioner.index; + +import org.apache.hudi.configuration.FlinkOptions; + +import org.apache.flink.configuration.Configuration; + +/** + * Factory to create a {@link PartitionedIndexBackend} used by the dynamic bucket assign function. + */ +public class PartitionedIndexBackendFactory { + private static final String ROCKSDB_BACKEND_TYPE = "rocksdb"; + + /** + * Creates the partitioned index backend used to look up and record {@code recordKey -> fileGroupId} + * mappings for the partitioned record level index. + * + * @param conf Flink write configuration + * @param isInsertOverwrite whether the write operation is an insert overwrite, in which case indexing is skipped + * @param bootstrapFilter filter for deciding whether a bootstrapped RLI record belongs to this task, + * used only by the metadata-table-backed backend + * @return partitioned index backend for record-key lookups scoped to a data partition + */ + public static PartitionedIndexBackend create( + Configuration conf, + boolean isInsertOverwrite, + RecordLevelIndexBackend.BootstrapFilter bootstrapFilter) { + if (isInsertOverwrite) { + return new DummyPartitionedIndexBackend(); + } + String backendType = conf.get(FlinkOptions.INDEX_RLI_BACKEND_TYPE); + if (ROCKSDB_BACKEND_TYPE.equalsIgnoreCase(backendType)) { + return new RocksDBPartitionedIndexBackend(conf.get(FlinkOptions.INDEX_RLI_CACHE_ROCKSDB_BASE_PATH)); Review Comment: 🤖 With `index.rli.backend.type=rocksdb`, `get()` only sees keys written via `update()` in the current run (RocksDBDAO wipes its dir on startup) and never consults the MDT RLI. On a non-empty table or after a restart, existing keys miss and get routed as inserts in DynamicBucketAssignFunction (L130-136), which can create a second file group for a key that already exists. The `index.rli.cache.rocksdb.base.path` description says this cache sits "in front of the metadata table" — is the MDT bootstrap/fallback still TODO before this backend is selectable? @danny0405 might want to weigh in. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/PartitionedIndexBackendFactory.java: ########## @@ -0,0 +1,54 @@ +/* + * 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.hudi.sink.partitioner.index; + +import org.apache.hudi.configuration.FlinkOptions; + +import org.apache.flink.configuration.Configuration; + +/** + * Factory to create a {@link PartitionedIndexBackend} used by the dynamic bucket assign function. + */ +public class PartitionedIndexBackendFactory { + private static final String ROCKSDB_BACKEND_TYPE = "rocksdb"; + + /** Review Comment: 🤖 nit: the string `"rocksdb"` is defined here but the valid values for `FlinkOptions.INDEX_RLI_BACKEND_TYPE` aren't visible at the call site. It might be clearer to keep this constant (or an enum) in `FlinkOptions` alongside the config key itself, so the two can't drift independently. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/RocksDBPartitionedIndexBackend.java: ########## @@ -0,0 +1,119 @@ +/* + * 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.hudi.sink.partitioner.index; + +import org.apache.hudi.common.util.StringUtils; +import org.apache.hudi.common.util.collection.Pair; +import org.apache.hudi.common.util.collection.RocksDBDAO; + +import lombok.extern.slf4j.Slf4j; +import org.rocksdb.RocksDB; + +import java.io.IOException; +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; + +/** + * An implementation of {@link PartitionedIndexBackend} based on RocksDB. + * + * <p>Each data partition is stored in its own RocksDB column family, so that a partition's mapping + * can be dropped as a unit. A partition is only registered as visible to {@link #get} in RocksDB's + * own {@code default} column family after its own column family has been created, so a lookup never + * sees a partition whose column family creation is still in flight. + * + * <p>This backend operates against an already-open RocksDB instance: it does not bootstrap partitions + * from the metadata table and does not implement TTL-based eviction of partitions. + */ +@Slf4j +public class RocksDBPartitionedIndexBackend implements PartitionedIndexBackend { + private static final String PARTITION_COLUMN_FAMILY_PREFIX = "pcf_"; + private static final String REGISTRY_COLUMN_FAMILY = StringUtils.fromUTF8Bytes(RocksDB.DEFAULT_COLUMN_FAMILY); + + private final RocksDBDAO rocksDBDAO; + + public RocksDBPartitionedIndexBackend(String rocksDbBasePath) { + this.rocksDBDAO = new RocksDBDAO("hudi-partitioned-index-backend", rocksDbBasePath, new ConcurrentHashMap<>(), true); + } + + @Override + public String get(String partitionPath, String recordKey) { + if (!isPartitionRegistered(partitionPath)) { + return null; + } + return this.rocksDBDAO.get(partitionColumnFamily(partitionPath), recordKey); + } + + @Override + public void update(String partitionPath, String recordKey, String fileId) { + String columnFamily = partitionColumnFamily(partitionPath); + if (!this.rocksDBDAO.columnFamilyExists(columnFamily)) { + this.rocksDBDAO.addColumnFamily(columnFamily); + } + this.rocksDBDAO.put(columnFamily, recordKey, fileId); + if (!isPartitionRegistered(partitionPath)) { + registerPartition(partitionPath); + } + } + + /** + * Returns whether the given partition's column family has been fully created and is safe to read from. + * + * @param partitionPath the partition path to check + */ + public boolean isPartitionRegistered(String partitionPath) { + Boolean registered = this.rocksDBDAO.get(REGISTRY_COLUMN_FAMILY, partitionPath); + return registered != null && registered; + } + + /** + * Lists the partitions currently registered in this backend. + */ + public List<String> listRegisteredPartitions() { + return this.rocksDBDAO.<Boolean>prefixSearch(REGISTRY_COLUMN_FAMILY, "") + .map(Pair::getKey) + .collect(Collectors.toList()); + } + + /** + * Drops a partition's column family and removes it from the registry. + * + * @param partitionPath the partition path to delete + */ + public void deletePartition(String partitionPath) { Review Comment: 🤖 `deletePartition` drops the column family before removing the registry entry — the reverse of the register-after-create ordering `update()` uses to keep `get()` safe. If a `get()` ever observed the intermediate state it would see the partition still registered but the CF handle already removed from RocksDBDAO's managedHandlesMap (null handle passed to RocksDB.get). Harmless while the assign operator is single-threaded, but would it be safer to unregister first, then drop the CF, to match the documented visibility invariant? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/index/RocksDBPartitionedIndexBackend.java: ########## @@ -0,0 +1,119 @@ +/* + * 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.hudi.sink.partitioner.index; + +import org.apache.hudi.common.util.StringUtils; +import org.apache.hudi.common.util.collection.Pair; +import org.apache.hudi.common.util.collection.RocksDBDAO; + +import lombok.extern.slf4j.Slf4j; +import org.rocksdb.RocksDB; + +import java.io.IOException; +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; + +/** + * An implementation of {@link PartitionedIndexBackend} based on RocksDB. + * + * <p>Each data partition is stored in its own RocksDB column family, so that a partition's mapping + * can be dropped as a unit. A partition is only registered as visible to {@link #get} in RocksDB's + * own {@code default} column family after its own column family has been created, so a lookup never + * sees a partition whose column family creation is still in flight. + * + * <p>This backend operates against an already-open RocksDB instance: it does not bootstrap partitions + * from the metadata table and does not implement TTL-based eviction of partitions. + */ +@Slf4j +public class RocksDBPartitionedIndexBackend implements PartitionedIndexBackend { + private static final String PARTITION_COLUMN_FAMILY_PREFIX = "pcf_"; + private static final String REGISTRY_COLUMN_FAMILY = StringUtils.fromUTF8Bytes(RocksDB.DEFAULT_COLUMN_FAMILY); + + private final RocksDBDAO rocksDBDAO; + + public RocksDBPartitionedIndexBackend(String rocksDbBasePath) { + this.rocksDBDAO = new RocksDBDAO("hudi-partitioned-index-backend", rocksDbBasePath, new ConcurrentHashMap<>(), true); + } + + @Override Review Comment: 🤖 nit: the parameter `rocksDbBasePath` uses lowercase `Db` while the class name spells it `RocksDB` (uppercase). Could you align to `rocksDBBasePath` for consistency? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
