@Test public void testAverageBytesPerRecordForEmptyCommitTimeLine() throws Exception { HoodieTimeline commitTimeLine = mock(HoodieTimeline.class); HoodieWriteConfig config = makeHoodieClientConfigBuilder().build(); when(commitTimeLine.empty()).thenReturn(true); long expectAvgSize = config.getCopyOnWriteRecordSizeEstimate(); long actualAvgSize = averageBytesPerRecord(commitTimeLine, config); assertEquals(expectAvgSize, actualAvgSize); }
protected static long averageBytesPerRecord(HoodieTimeline commitTimeline, HoodieWriteConfig hoodieWriteConfig) { long avgSize = hoodieWriteConfig.getCopyOnWriteRecordSizeEstimate(); long fileSizeThreshold = (long) (hoodieWriteConfig.getRecordSizeEstimationThreshold() * hoodieWriteConfig.getParquetSmallFileLimit()); try { if (!commitTimeline.empty()) { Iterator<HoodieInstant> instants = commitTimeline.getReverseOrderedInstants().iterator(); while (instants.hasNext()) { HoodieInstant instant = instants.next(); HoodieCommitMetadata commitMetadata = HoodieCommitMetadata .fromBytes(commitTimeline.getInstantDetails(instant).get(), HoodieCommitMetadata.class); long totalBytesWritten = commitMetadata.fetchTotalBytesWritten(); long totalRecordsWritten = commitMetadata.fetchTotalRecordsWritten(); if (totalBytesWritten > fileSizeThreshold && totalRecordsWritten > 0) { avgSize = (long) Math.ceil((1.0 * totalBytesWritten) / totalRecordsWritten); break; } } } } catch (Throwable t) { LOG.error("Error trying to compute average bytes/record ", t); } return avgSize; }
UpsertPartitioner extends Partitioner { protected static long averageBytesPerRecord(HoodieTimeline commitTimeline, HoodieWriteConfig hoodieWriteConfig) { long avgSize = hoodieWriteConfig.getCopyOnWriteRecordSizeEstimate(); long fileSizeThreshold = (long) (hoodieWriteConfig.getRecordSizeEstimationThreshold() * hoodieWriteConfig.getParquetSmallFileLimit()); try { if (!commitTimeline.empty()) { Iterator<HoodieInstant> instants = commitTimeline.getReverseOrderedInstants().iterator(); while (instants.hasNext()) { HoodieInstant instant = instants.next(); HoodieCommitMetadata commitMetadata = HoodieCommitMetadata .fromBytes(commitTimeline.getInstantDetails(instant).get(), HoodieCommitMetadata.class); long totalBytesWritten = commitMetadata.fetchTotalBytesWritten(); long totalRecordsWritten = commitMetadata.fetchTotalRecordsWritten(); if (totalBytesWritten > fileSizeThreshold && totalRecordsWritten > 0) { avgSize = (long) Math.ceil((1.0 * totalBytesWritten) / totalRecordsWritten); break; } } } } catch (Throwable t) { LOG.error("Error trying to compute average bytes/record ", t); } return avgSize; } }
UpsertPartitioner extends Partitioner { protected static long averageBytesPerRecord(HoodieTimeline commitTimeline, HoodieWriteConfig hoodieWriteConfig) { long avgSize = hoodieWriteConfig.getCopyOnWriteRecordSizeEstimate(); long fileSizeThreshold = (long) (hoodieWriteConfig.getRecordSizeEstimationThreshold() * hoodieWriteConfig.getParquetSmallFileLimit()); try { if (!commitTimeline.empty()) { Iterator<HoodieInstant> instants = commitTimeline.getReverseOrderedInstants().iterator(); while (instants.hasNext()) { HoodieInstant instant = instants.next(); HoodieCommitMetadata commitMetadata = HoodieCommitMetadata .fromBytes(commitTimeline.getInstantDetails(instant).get(), HoodieCommitMetadata.class); long totalBytesWritten = commitMetadata.fetchTotalBytesWritten(); long totalRecordsWritten = commitMetadata.fetchTotalRecordsWritten(); if (totalBytesWritten > fileSizeThreshold && totalRecordsWritten > 0) { avgSize = (long) Math.ceil((1.0 * totalBytesWritten) / totalRecordsWritten); break; } } } } catch (Throwable t) { LOG.error("Error trying to compute average bytes/record ", t); } return avgSize; } UpsertPartitioner(WorkloadProfile profile, HoodieEngineContext context, HoodieTable table,
      HoodieWriteConfig config); }
UpsertPartitioner extends Partitioner { protected static long averageBytesPerRecord(HoodieTimeline commitTimeline, HoodieWriteConfig hoodieWriteConfig) { long avgSize = hoodieWriteConfig.getCopyOnWriteRecordSizeEstimate(); long fileSizeThreshold = (long) (hoodieWriteConfig.getRecordSizeEstimationThreshold() * hoodieWriteConfig.getParquetSmallFileLimit()); try { if (!commitTimeline.empty()) { Iterator<HoodieInstant> instants = commitTimeline.getReverseOrderedInstants().iterator(); while (instants.hasNext()) { HoodieInstant instant = instants.next(); HoodieCommitMetadata commitMetadata = HoodieCommitMetadata .fromBytes(commitTimeline.getInstantDetails(instant).get(), HoodieCommitMetadata.class); long totalBytesWritten = commitMetadata.fetchTotalBytesWritten(); long totalRecordsWritten = commitMetadata.fetchTotalRecordsWritten(); if (totalBytesWritten > fileSizeThreshold && totalRecordsWritten > 0) { avgSize = (long) Math.ceil((1.0 * totalBytesWritten) / totalRecordsWritten); break; } } } } catch (Throwable t) { LOG.error("Error trying to compute average bytes/record ", t); } return avgSize; } UpsertPartitioner(WorkloadProfile profile, HoodieEngineContext context, HoodieTable table,
      HoodieWriteConfig config); BucketInfo getBucketInfo(int bucketNumber); List<InsertBucketCumulativeWeightPair> getInsertBuckets(String partitionPath); @Override int numPartitions(); @Override int getPartition(Object key); }
UpsertPartitioner extends Partitioner { protected static long averageBytesPerRecord(HoodieTimeline commitTimeline, HoodieWriteConfig hoodieWriteConfig) { long avgSize = hoodieWriteConfig.getCopyOnWriteRecordSizeEstimate(); long fileSizeThreshold = (long) (hoodieWriteConfig.getRecordSizeEstimationThreshold() * hoodieWriteConfig.getParquetSmallFileLimit()); try { if (!commitTimeline.empty()) { Iterator<HoodieInstant> instants = commitTimeline.getReverseOrderedInstants().iterator(); while (instants.hasNext()) { HoodieInstant instant = instants.next(); HoodieCommitMetadata commitMetadata = HoodieCommitMetadata .fromBytes(commitTimeline.getInstantDetails(instant).get(), HoodieCommitMetadata.class); long totalBytesWritten = commitMetadata.fetchTotalBytesWritten(); long totalRecordsWritten = commitMetadata.fetchTotalRecordsWritten(); if (totalBytesWritten > fileSizeThreshold && totalRecordsWritten > 0) { avgSize = (long) Math.ceil((1.0 * totalBytesWritten) / totalRecordsWritten); break; } } } } catch (Throwable t) { LOG.error("Error trying to compute average bytes/record ", t); } return avgSize; } UpsertPartitioner(WorkloadProfile profile, HoodieEngineContext context, HoodieTable table,
      HoodieWriteConfig config); BucketInfo getBucketInfo(int bucketNumber); List<InsertBucketCumulativeWeightPair> getInsertBuckets(String partitionPath); @Override int numPartitions(); @Override int getPartition(Object key);  }