mcvsubbu commented on a change in pull request #7062:
URL: https://github.com/apache/incubator-pinot/pull/7062#discussion_r656598132
##########
File path:
pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/HadoopSegmentPreprocessingJob.java
##########
@@ -215,123 +158,84 @@ private void fetchPartitioningConfig() {
_partitionFunction =
segmentPartitionConfig.getFunctionName(_partitionColumn);
}
} else {
- _logger.info("Segment partition config is null for table: {}",
_tableConfig.getTableName());
+ LOGGER.info("Segment partition config is null for table: {}",
_tableConfig.getTableName());
}
}
private void fetchSortingConfig() {
- // Fetch sorting info from table config.
+ if (!_preprocessingOperations.contains("sort")) {
+ LOGGER.info("Sorting is disabled.");
+ return;
+ }
+ // Fetch sorting info from table config first.
+ List<String> sortingColumns = new ArrayList<>();
+ List<FieldConfig> fieldConfigs = _tableConfig.getFieldConfigList();
+ if (fieldConfigs != null && !fieldConfigs.isEmpty()) {
+ for (FieldConfig fieldConfig : fieldConfigs) {
+ if (fieldConfig.getIndexType() == FieldConfig.IndexType.SORTED) {
+ sortingColumns.add(fieldConfig.getName());
+ }
+ }
+ }
+ if (!sortingColumns.isEmpty()) {
+ Preconditions.checkArgument(sortingColumns.size() == 1, "There should be
at most 1 sorted column in the table.");
+ _sortingColumn = sortingColumns.get(0);
+ return;
+ }
+
+ // There is no sorted column specified in field configs, try to find
sorted column from indexing config.
IndexingConfig indexingConfig = _tableConfig.getIndexingConfig();
List<String> sortedColumns = indexingConfig.getSortedColumn();
if (sortedColumns != null) {
Preconditions.checkArgument(sortedColumns.size() <= 1, "There should be
at most 1 sorted column in the table.");
if (sortedColumns.size() == 1) {
- _sortedColumn = sortedColumns.get(0);
+ _sortingColumn = sortedColumns.get(0);
+ FieldSpec fieldSpec =
_pinotTableSchema.getFieldSpecFor(_sortingColumn);
+ Preconditions.checkState(fieldSpec != null, "Failed to find sorting
column: {} in the schema", _sortingColumn);
+ Preconditions
+ .checkState(fieldSpec.isSingleValueField(), "Cannot sort on
multi-value column: %s", _sortingColumn);
+ _sortingColumnType = fieldSpec.getDataType();
+ Preconditions.checkState(_sortingColumnType !=
FieldSpec.DataType.BYTES, "Cannot sort on BYTES column: %s",
+ _sortingColumn);
+ LOGGER.info("Sorting the data with column: {} of type: {}",
_sortingColumn, _sortingColumnType);
}
}
}
private void fetchResizingConfig() {
+ if (!_preprocessingOperations.contains("resize")) {
+ LOGGER.info("Resizing is disabled.");
+ return;
+ }
TableCustomConfig tableCustomConfig = _tableConfig.getCustomConfig();
if (tableCustomConfig == null) {
_numOutputFiles = 0;
return;
}
Map<String, String> customConfigsMap =
tableCustomConfig.getCustomConfigs();
- if (customConfigsMap != null &&
customConfigsMap.containsKey(InternalConfigConstants.PREPROCESS_NUM_FILES)) {
- _numOutputFiles =
Integer.parseInt(customConfigsMap.get(InternalConfigConstants.PREPROCESS_NUM_FILES));
+ if (customConfigsMap != null &&
customConfigsMap.containsKey(InternalConfigConstants.PREPROCESSING_NUM_REDUCERS))
{
+ _numOutputFiles =
Integer.parseInt(customConfigsMap.get(InternalConfigConstants.PREPROCESSING_NUM_REDUCERS));
Preconditions.checkState(_numOutputFiles > 0, String
- .format("The value of %s should be positive! Current value: %s",
InternalConfigConstants.PREPROCESS_NUM_FILES,
- _numOutputFiles));
+ .format("The value of %s should be positive! Current value: %s",
+ InternalConfigConstants.PREPROCESSING_NUM_REDUCERS,
_numOutputFiles));
} else {
_numOutputFiles = 0;
}
- }
- private void setMaxNumRecordsConfigIfSpecified(Job job) {
- TableCustomConfig tableCustomConfig = _tableConfig.getCustomConfig();
- if (tableCustomConfig == null) {
- return;
- }
- Map<String, String> customConfigsMap =
tableCustomConfig.getCustomConfigs();
if (customConfigsMap != null && customConfigsMap
- .containsKey(InternalConfigConstants.PARTITION_MAX_RECORDS_PER_FILE)) {
+
.containsKey(InternalConfigConstants.PREPROCESSING_MAX_NUM_RECORDS_PER_FILE)) {
int maxNumRecords =
-
Integer.parseInt(customConfigsMap.get(InternalConfigConstants.PARTITION_MAX_RECORDS_PER_FILE));
+
Integer.parseInt(customConfigsMap.get(InternalConfigConstants.PREPROCESSING_MAX_NUM_RECORDS_PER_FILE));
Preconditions.checkArgument(maxNumRecords > 0,
- "The value of " +
InternalConfigConstants.PARTITION_MAX_RECORDS_PER_FILE
+ "The value of " +
InternalConfigConstants.PREPROCESSING_MAX_NUM_RECORDS_PER_FILE
Review comment:
nit: The TODO is to add some logic so that we don't have too many small
files. Please fix the comment.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
