nsivabalan commented on code in PR #18409: URL: https://github.com/apache/hudi/pull/18409#discussion_r3034567605
########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieFileWriteHandle.java: ########## @@ -0,0 +1,113 @@ +/* + * 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.io; + +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.common.engine.TaskContextSupplier; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieRecordPayload; +import org.apache.hudi.common.model.IOType; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.exception.HoodieInsertException; +import org.apache.hudi.execution.ParquetFileMetaToWriteStatusConvertor; +import org.apache.hudi.storage.StoragePath; +import org.apache.hudi.table.HoodieTable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.hudi.execution.ParquetFileMetaToWriteStatusConvertor.PREV_COMMIT; +import static org.apache.hudi.execution.ParquetFileMetaToWriteStatusConvertor.TIME_TAKEN; + +/** + * Write handle that is used to work on top of files rather than on individual records. + */ +public class HoodieFileWriteHandle<T extends HoodieRecordPayload, I, K, O> extends HoodieWriteHandle<T, I, K, O> { + + private static final Logger LOG = LoggerFactory.getLogger(HoodieFileWriteHandle.class); + private final StoragePath path; + private final String prevCommit; + + public HoodieFileWriteHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable, + String partitionPath, String fileId, TaskContextSupplier taskContextSupplier, + StoragePath oldFilePath) { + super(config, instantTime, partitionPath, fileId, hoodieTable, taskContextSupplier, true); + + // Output file path. + this.path = makeNewPath(partitionPath); + // Get the prev commit from existing or old file. + this.prevCommit = FSUtils.getCommitTime(oldFilePath.getName()); + + // Create inProgress marker file + createMarkerFile(partitionPath, path.getName()); + LOG.info("New HoodieFileWriteHandle for partition :" + partitionPath + " with fileId " + fileId); + } + + /** + * Complete writing of the file by creating the success marker file. + * @return WriteStatuses, ideally it will be only one object. + */ + @Override + public List<WriteStatus> close() { + LOG.info("Closing the file " + this.fileId + " as we are done with the file."); Review Comment: Merge handle logs something like ``` log.info("MergeHandle for partitionPath {} fileID {}, took {} ms.", stat.getPartitionPath(), stat.getFileId(), runtimeStats.getTotalUpsertTime()); ``` We should try to follow the same pattern. also, lets avoid 2 info logs at L 73 and 80. can we keep it to just 1 log ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieFileWriteHandle.java: ########## @@ -0,0 +1,113 @@ +/* + * 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.io; + +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.common.engine.TaskContextSupplier; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieRecordPayload; +import org.apache.hudi.common.model.IOType; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.exception.HoodieInsertException; +import org.apache.hudi.execution.ParquetFileMetaToWriteStatusConvertor; +import org.apache.hudi.storage.StoragePath; +import org.apache.hudi.table.HoodieTable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.hudi.execution.ParquetFileMetaToWriteStatusConvertor.PREV_COMMIT; +import static org.apache.hudi.execution.ParquetFileMetaToWriteStatusConvertor.TIME_TAKEN; + +/** + * Write handle that is used to work on top of files rather than on individual records. + */ +public class HoodieFileWriteHandle<T extends HoodieRecordPayload, I, K, O> extends HoodieWriteHandle<T, I, K, O> { + + private static final Logger LOG = LoggerFactory.getLogger(HoodieFileWriteHandle.class); + private final StoragePath path; + private final String prevCommit; + + public HoodieFileWriteHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable, + String partitionPath, String fileId, TaskContextSupplier taskContextSupplier, + StoragePath oldFilePath) { + super(config, instantTime, partitionPath, fileId, hoodieTable, taskContextSupplier, true); + + // Output file path. + this.path = makeNewPath(partitionPath); + // Get the prev commit from existing or old file. + this.prevCommit = FSUtils.getCommitTime(oldFilePath.getName()); + + // Create inProgress marker file + createMarkerFile(partitionPath, path.getName()); + LOG.info("New HoodieFileWriteHandle for partition :" + partitionPath + " with fileId " + fileId); + } + + /** + * Complete writing of the file by creating the success marker file. Review Comment: comments needs to be fixed. we don't have any completion marker. ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/ParquetFileMetaToWriteStatusConvertor.java: ########## @@ -0,0 +1,103 @@ +/* + * 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.execution; + +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieRecordPayload; +import org.apache.hudi.common.model.HoodieWriteStat; +import org.apache.hudi.common.util.ParquetUtils; +import org.apache.hudi.common.util.ReflectionUtils; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.storage.HoodieStorage; +import org.apache.hudi.storage.StoragePath; +import org.apache.hudi.table.HoodieTable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.Map; + +/** + * This class is mainly used by the ParquetToolsExecutionStrategy to generate WriteStatus classes. + */ +public class ParquetFileMetaToWriteStatusConvertor<T extends HoodieRecordPayload, I, K, O> { + + private static final Logger LOG = LoggerFactory.getLogger(ParquetFileMetaToWriteStatusConvertor.class); + private final HoodieTable<T, I, K, O> hoodieTable; + private final HoodieWriteConfig writeConfig; + private final HoodieStorage storage; + private final ParquetUtils parquetUtils; + public static final String PREV_COMMIT = "prevCommit"; + public static final String TIME_TAKEN = "timeTaken"; Review Comment: totalCreateTime ########## hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/ParquetToolsTestExecutionStrategy.java: ########## @@ -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. + */ + +package org.apache.hudi.client; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.FileUtil; +import org.apache.hadoop.fs.Path; +import org.apache.hudi.client.clustering.run.strategy.ParquetToolsExecutionStrategy; +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.model.HoodieRecordPayload; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.exception.HoodieIOException; +import org.apache.hudi.table.HoodieTable; + +import java.io.IOException; + +/** + * Test execution strategy for testing the skeleton of the ParquetToolsExecutionStrategy. + * It creates a copy of the original file with a different commit timestamp. + */ +public class ParquetToolsTestExecutionStrategy<T extends HoodieRecordPayload<T>> Review Comment: where is this being used? ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieFileWriteHandle.java: ########## @@ -0,0 +1,113 @@ +/* + * 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.io; + +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.common.engine.TaskContextSupplier; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieRecordPayload; +import org.apache.hudi.common.model.IOType; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.exception.HoodieInsertException; +import org.apache.hudi.execution.ParquetFileMetaToWriteStatusConvertor; +import org.apache.hudi.storage.StoragePath; +import org.apache.hudi.table.HoodieTable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.hudi.execution.ParquetFileMetaToWriteStatusConvertor.PREV_COMMIT; +import static org.apache.hudi.execution.ParquetFileMetaToWriteStatusConvertor.TIME_TAKEN; + +/** + * Write handle that is used to work on top of files rather than on individual records. + */ +public class HoodieFileWriteHandle<T extends HoodieRecordPayload, I, K, O> extends HoodieWriteHandle<T, I, K, O> { + + private static final Logger LOG = LoggerFactory.getLogger(HoodieFileWriteHandle.class); + private final StoragePath path; + private final String prevCommit; + + public HoodieFileWriteHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable, + String partitionPath, String fileId, TaskContextSupplier taskContextSupplier, + StoragePath oldFilePath) { + super(config, instantTime, partitionPath, fileId, hoodieTable, taskContextSupplier, true); + + // Output file path. + this.path = makeNewPath(partitionPath); + // Get the prev commit from existing or old file. + this.prevCommit = FSUtils.getCommitTime(oldFilePath.getName()); + + // Create inProgress marker file + createMarkerFile(partitionPath, path.getName()); + LOG.info("New HoodieFileWriteHandle for partition :" + partitionPath + " with fileId " + fileId); + } + + /** + * Complete writing of the file by creating the success marker file. + * @return WriteStatuses, ideally it will be only one object. + */ + @Override + public List<WriteStatus> close() { + LOG.info("Closing the file " + this.fileId + " as we are done with the file."); + try { + Map<String, Object> executionConfigs = new HashMap<>(); + executionConfigs.put(PREV_COMMIT, prevCommit); + executionConfigs.put(TIME_TAKEN, timer.endTimer()); + + this.writeStatus = generateWriteStatus(path.toString(), partitionPath, executionConfigs); + LOG.info(String.format("HoodieFileWriteHandle for partitionPath %s fileID %s, took %d ms.", + writeStatus.getStat().getPartitionPath(), writeStatus.getStat().getFileId(), + writeStatus.getStat().getRuntimeStats().getTotalCreateTime())); + + return Collections.singletonList(writeStatus); + } catch (IOException e) { + throw new HoodieInsertException("Failed to close the HoodieFileWriteHandle for path " + path, e); + } + } + + /** + * Given a parquet file it generates WriteStatus object for a parquet file. + * @param outputFile parquet file + * @param partitionPath partition information of the parquet file + * @param executionConfigs Some configs collected during execution. + * @return WriteStatus object. + * @throws IOException + */ + protected WriteStatus generateWriteStatus( + String outputFile, String partitionPath, Map<String, Object> executionConfigs) throws IOException { + ParquetFileMetaToWriteStatusConvertor convertor = + new ParquetFileMetaToWriteStatusConvertor(hoodieTable, config); + return convertor.convert(outputFile, partitionPath, executionConfigs); Review Comment: we could directly do ``` return new ParquetFileMetaToWriteStatusConvertor(hoodieTable, config).convert(outputFile, partitionPath, executionConfigs) ``` ########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/execution/ParquetFileMetaToWriteStatusConvertor.java: ########## @@ -0,0 +1,103 @@ +/* + * 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.execution; + +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieRecordPayload; +import org.apache.hudi.common.model.HoodieWriteStat; +import org.apache.hudi.common.util.ParquetUtils; +import org.apache.hudi.common.util.ReflectionUtils; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.storage.HoodieStorage; +import org.apache.hudi.storage.StoragePath; +import org.apache.hudi.table.HoodieTable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.Map; + +/** + * This class is mainly used by the ParquetToolsExecutionStrategy to generate WriteStatus classes. + */ +public class ParquetFileMetaToWriteStatusConvertor<T extends HoodieRecordPayload, I, K, O> { + + private static final Logger LOG = LoggerFactory.getLogger(ParquetFileMetaToWriteStatusConvertor.class); + private final HoodieTable<T, I, K, O> hoodieTable; + private final HoodieWriteConfig writeConfig; + private final HoodieStorage storage; + private final ParquetUtils parquetUtils; + public static final String PREV_COMMIT = "prevCommit"; + public static final String TIME_TAKEN = "timeTaken"; + + public ParquetFileMetaToWriteStatusConvertor(HoodieTable<T, I, K, O> hoodieTable, HoodieWriteConfig writeConfig) { + this.hoodieTable = hoodieTable; + this.writeConfig = writeConfig; + this.storage = this.hoodieTable.getStorage(); + this.parquetUtils = new ParquetUtils(); + } + + /** + * This method generates writeStatus object from parquet file. + */ + public WriteStatus convert(String parquetFile, String partitionPath, + Map<String, Object> executionConfigs) throws IOException { + LOG.info("Creating write status for parquet file " + parquetFile); + WriteStatus writeStatus = (WriteStatus) ReflectionUtils.loadClass(this.writeConfig.getWriteStatusClassName(), + !this.hoodieTable.getIndex().isImplicitWithStorage(), this.writeConfig.getWriteStatusFailureFraction()); + StoragePath parquetFilePath = new StoragePath(parquetFile); + writeStatus.setFileId(FSUtils.getFileId(parquetFilePath.getName())); + writeStatus.setPartitionPath(partitionPath); + generateHoodieWriteStat(writeStatus, parquetFilePath, executionConfigs); + return writeStatus; + } + + /** + * This method generates HoodieWriteStat object and set it as part of WriteStatus object. + */ + private void generateHoodieWriteStat( + WriteStatus writeStatus, StoragePath parquetFilePath, Map<String, Object> executionConfigs) throws IOException { + HoodieWriteStat stat = new HoodieWriteStat(); + + // Set row count + long rowCount = parquetUtils.getRowCount(storage, parquetFilePath); + stat.setNumWrites(rowCount); + stat.setNumInserts(rowCount); + + // Set runtime stats + HoodieWriteStat.RuntimeStats runtimeStats = new HoodieWriteStat.RuntimeStats(); + runtimeStats.setTotalCreateTime(((Number) executionConfigs.get(TIME_TAKEN)).longValue()); + stat.setRuntimeStats(runtimeStats); + + // File size + long fileSize = storage.getPathInfo(parquetFilePath).getLength(); + stat.setFileSizeInBytes(fileSize); + stat.setTotalWriteBytes(fileSize); + + stat.setFileId(writeStatus.getFileId()); + stat.setPartitionPath(writeStatus.getPartitionPath()); + stat.setPath(new StoragePath(writeConfig.getBasePath()), parquetFilePath); + stat.setTotalWriteErrors(writeStatus.getTotalErrorRecords()); Review Comment: what does this even mean. we create a new `WriteStatus` at L 63. So, we could never has any error records right? -- 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]
