DRILL-7223: Create an option to control timeout for REFRESH METADATA

exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java

@@ -354,6 +354,10 @@ private ExecConstants() {
  "enables statistics usage for varchar and decimal data types. Default is unset, i.e. empty string. " +
  "Allowed values: 'true', 'false', '' (empty string)."), "true", "false", "");
 
+ public static final String PARQUET_REFRESH_TIMEOUT = "store.parquet.refresh_timeout_per_runnable_in_msec";
+ public static final LongValidator PARQUET_REFRESH_TIMEOUT_VALIDATOR = new LongValidator(PARQUET_REFRESH_TIMEOUT,
+ new OptionDescription("Sets a timeout (in msec) for REFRESH TABLE METADATA processing of a single subdirectory"));
+
  public static final String PARQUET_PAGEREADER_ASYNC = "store.parquet.reader.pagereader.async";
  public static final OptionValidator PARQUET_PAGEREADER_ASYNC_VALIDATOR = new BooleanValidator(PARQUET_PAGEREADER_ASYNC,
  new OptionDescription("Enable the asynchronous page reader. This pipelines the reading of data from disk for high performance."));

exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java

@@ -174,6 +174,7 @@ public static CaseInsensitiveMap<OptionDefinition> createDefaultOptionDefinition
  new OptionDefinition(ExecConstants.PARQUET_PAGEREADER_USE_FADVISE_VALIDATOR),
  new OptionDefinition(ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP_VALIDATOR),
  new OptionDefinition(ExecConstants.PARQUET_READER_STRINGS_SIGNED_MIN_MAX_VALIDATOR),
+ new OptionDefinition(ExecConstants.PARQUET_REFRESH_TIMEOUT_VALIDATOR),
  new OptionDefinition(ExecConstants.PARQUET_FLAT_READER_BULK_VALIDATOR),
  new OptionDefinition(ExecConstants.PARQUET_FLAT_BATCH_NUM_RECORDS_VALIDATOR, new OptionMetaData(OptionValue.AccessibleScopes.SYSTEM_AND_SESSION, true, true)),
  new OptionDefinition(ExecConstants.PARQUET_FLAT_BATCH_MEMORY_SIZE_VALIDATOR, new OptionMetaData(OptionValue.AccessibleScopes.SYSTEM_AND_SESSION, true, true)),

exec/java-exec/src/main/java/org/apache/drill/exec/store/TimedCallable.java

@@ -166,14 +166,13 @@ public final V call() throws Exception {
  throw e;
  } finally {
  long time = System.nanoTime() - start;
- if (logger.isWarnEnabled()) {
- long timeMillis = TimeUnit.MILLISECONDS.convert(time, TimeUnit.NANOSECONDS);
- if (timeMillis > TIMEOUT_PER_RUNNABLE_IN_MSECS) {
- logger.warn("Task '{}' execution time {} ms exceeds timeout {} ms.", this, timeMillis, TIMEOUT_PER_RUNNABLE_IN_MSECS);
- } else {
- logger.debug("Task '{}' execution time is {} ms", this, timeMillis);
- }
+ long timeMillis = TimeUnit.MILLISECONDS.convert(time, TimeUnit.NANOSECONDS);
+ if (timeMillis > TIMEOUT_PER_RUNNABLE_IN_MSECS) {
+ logger.warn("Task '{}' execution time {} ms exceeds timeout {} ms.", this, timeMillis, TIMEOUT_PER_RUNNABLE_IN_MSECS);
+ } else {
+ logger.debug("Task '{}' execution time is {} ms", this, timeMillis);
  }
+
  executionTime = time;
  }
  }
@@ -188,6 +187,23 @@ private long getExecutionTime(TimeUnit unit) {
  return unit.convert(executionTime, TimeUnit.NANOSECONDS);
  }
 
+ /**
+ * Execute the list of runnables with the given parallelization. At end, return values and report completion time
+ * stats to provided logger. Each runnable is allowed a certain timeout. If the timeout exceeds, existing/pending
+ * tasks will be cancelled and a {@link UserException} is thrown.
+ * @param activity Name of activity for reporting in logger.
+ * @param logger The logger to use to report results.
+ * @param tasks List of callable that should be executed and timed. If this list has one item, task will be
+ * completed in-thread. Each callable must handle {@link InterruptedException}s.
+ * @param parallelism The number of threads that should be run to complete this task.
+ * @param timeout if bigger than zero, set the timeout per runnable (in msec)
+ * @return The list of outcome objects.
+ * @throws IOException All exceptions are coerced to IOException since this was build for storage system tasks initially.
+ */
+ public static <V> List<V> run(final String activity, final Logger logger, final List<TimedCallable<V>> tasks, int parallelism, long timeout) throws IOException {
+ TIMEOUT_PER_RUNNABLE_IN_MSECS = timeout;
+ return run(activity, logger, tasks, parallelism);
+ }
 
  /**
  * Execute the list of runnables with the given parallelization. At end, return values and report completion time

exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetReaderConfig.java

@@ -50,6 +50,7 @@ public class ParquetReaderConfig {
  private boolean enableTimeReadCounter = false;
  private boolean autoCorrectCorruptedDates = true;
  private boolean enableStringsSignedMinMax = false;
+ private long timeoutPerRunnableInMsec = 15_000;
 
  public static ParquetReaderConfig.Builder builder() {
  return new ParquetReaderConfig.Builder();
@@ -64,12 +65,15 @@ public ParquetReaderConfig(@JsonProperty("enableBytesReadCounter") Boolean enabl
  @JsonProperty("enableBytesTotalCounter") Boolean enableBytesTotalCounter,
  @JsonProperty("enableTimeReadCounter") Boolean enableTimeReadCounter,
  @JsonProperty("autoCorrectCorruptedDates") Boolean autoCorrectCorruptedDates,
- @JsonProperty("enableStringsSignedMinMax") Boolean enableStringsSignedMinMax) {
+ @JsonProperty("enableStringsSignedMinMax") Boolean enableStringsSignedMinMax,
+ @JsonProperty("timeoutPerRunnableInMsec") Long timeoutPerRunnableInMsec) {
  this.enableBytesReadCounter = enableBytesReadCounter == null ? this.enableBytesReadCounter : enableBytesReadCounter;
  this.enableBytesTotalCounter = enableBytesTotalCounter == null ? this.enableBytesTotalCounter : enableBytesTotalCounter;
  this.enableTimeReadCounter = enableTimeReadCounter == null ? this.enableTimeReadCounter : enableTimeReadCounter;
  this.autoCorrectCorruptedDates = autoCorrectCorruptedDates == null ? this.autoCorrectCorruptedDates : autoCorrectCorruptedDates;
  this.enableStringsSignedMinMax = enableStringsSignedMinMax == null ? this.enableStringsSignedMinMax : enableStringsSignedMinMax;
+ this.timeoutPerRunnableInMsec = timeoutPerRunnableInMsec == null || Long.valueOf(timeoutPerRunnableInMsec) <= 0 ? // zero means: use default
+ this.timeoutPerRunnableInMsec : timeoutPerRunnableInMsec;
  }
 
  private ParquetReaderConfig() { }
@@ -99,6 +103,9 @@ public boolean enableStringsSignedMinMax() {
  return enableStringsSignedMinMax;
  }
 
+ @JsonProperty("timeoutPerRunnableInMsec")
+ public long timeoutPerRunnableInMsec() { return timeoutPerRunnableInMsec; }
+
  public ParquetReadOptions toReadOptions() {
  return ParquetReadOptions.builder()
  .useSignedStringMinMax(enableStringsSignedMinMax)
@@ -119,7 +126,8 @@ public int hashCode() {
  enableBytesTotalCounter,
  enableTimeReadCounter,
  autoCorrectCorruptedDates,
- enableStringsSignedMinMax);
+ enableStringsSignedMinMax,
+ timeoutPerRunnableInMsec);
  }
 
  @Override
@@ -135,7 +143,8 @@ public boolean equals(Object o) {
  && enableBytesTotalCounter == that.enableBytesTotalCounter
  && enableTimeReadCounter == that.enableTimeReadCounter
  && autoCorrectCorruptedDates == that.autoCorrectCorruptedDates
- && enableStringsSignedMinMax == that.enableStringsSignedMinMax;
+ && enableStringsSignedMinMax == that.enableStringsSignedMinMax
+ && timeoutPerRunnableInMsec == that.timeoutPerRunnableInMsec;
  }
 
  @Override
@@ -146,6 +155,7 @@ public String toString() {
  + ", enableTimeReadCounter=" + enableTimeReadCounter
  + ", autoCorrectCorruptedDates=" + autoCorrectCorruptedDates
  + ", enableStringsSignedMinMax=" + enableStringsSignedMinMax
+ + ", timeoutPerRunnableInMsec=" + timeoutPerRunnableInMsec
  + '}';
  }
 
@@ -188,10 +198,12 @@ public ParquetReaderConfig build() {
 
  // last assign values from session options, session options have higher priority than other configurations
  if (options != null) {
- String option = options.getOption(ExecConstants.PARQUET_READER_STRINGS_SIGNED_MIN_MAX_VALIDATOR);
- if (!option.isEmpty()) {
- readerConfig.enableStringsSignedMinMax = Boolean.valueOf(option);
+ String optionSignedMinMax = options.getOption(ExecConstants.PARQUET_READER_STRINGS_SIGNED_MIN_MAX_VALIDATOR);
+ if (!optionSignedMinMax.isEmpty()) {
+ readerConfig.enableStringsSignedMinMax = Boolean.valueOf(optionSignedMinMax);
  }
+ Long optionTimeout = options.getOption(ExecConstants.PARQUET_REFRESH_TIMEOUT_VALIDATOR);
+ readerConfig.timeoutPerRunnableInMsec = Long.valueOf(optionTimeout);
  }
 
  return readerConfig;

exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata.java

@@ -444,8 +444,7 @@ private ParquetTableMetadata_v4 getParquetTableMetadata(Map<FileStatus, FileSyst
  private List<ParquetFileAndRowCountMetadata> getParquetFileMetadata_v4(ParquetTableMetadata_v4 parquetTableMetadata_v4, Map<FileStatus, FileSystem> fileStatusMap, boolean allColumnsInteresting, Set<String> columnSet) throws IOException {
  return TimedCallable.run("Fetch parquet metadata", logger,
  Collectors.toList(fileStatusMap,
- (fileStatus, fileSystem) -> new MetadataGatherer(parquetTableMetadata_v4, fileStatus, fileSystem, allColumnsInteresting, columnSet)),
- 16
+ (fileStatus, fileSystem) -> new MetadataGatherer(parquetTableMetadata_v4, fileStatus, fileSystem, allColumnsInteresting, columnSet)), 16, readerConfig.timeoutPerRunnableInMsec()
  );
  }
 

exec/java-exec/src/main/resources/drill-module.conf

@@ -629,6 +629,7 @@ drill.exec.options: {
  store.parquet.reader.columnreader.async: false,
  store.parquet.reader.int96_as_timestamp: false,
  store.parquet.reader.strings_signed_min_max: "",
+ store.parquet.refresh_timeout_per_runnable_in_msec: 15000,
  store.parquet.reader.pagereader.async: true,
  store.parquet.reader.pagereader.bufferedread: true,
  store.parquet.reader.pagereader.buffersize: 1048576,

exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/TestParquetReaderConfig.java

@@ -45,7 +45,7 @@ public void testDefaultsDeserialization() throws Exception {
 
  // change the default: set autoCorrectCorruptedDates to false
  // keep the default: set enableStringsSignedMinMax to false
- readerConfig = new ParquetReaderConfig(false, false, false, false, false);
+ readerConfig = new ParquetReaderConfig(false, false, false, false, false, 0L);
 
  value = mapper.writeValueAsString(readerConfig);
  assertEquals("{\"autoCorrectCorruptedDates\":false}", value);
@@ -79,12 +79,12 @@ public void testAddConfigToConf() {
  @Test
  public void testReadOptions() {
  // set enableStringsSignedMinMax to true
- ParquetReaderConfig readerConfig = new ParquetReaderConfig(false, false, false, true, true);
+ ParquetReaderConfig readerConfig = new ParquetReaderConfig(false, false, false, true, true, 12345L);
  ParquetReadOptions readOptions = readerConfig.toReadOptions();
  assertTrue(readOptions.useSignedStringMinMax());
 
  // set enableStringsSignedMinMax to false
- readerConfig = new ParquetReaderConfig(false, false, false, true, false);
+ readerConfig = new ParquetReaderConfig(false, false, false, true, false, 12345L);
  readOptions = readerConfig.toReadOptions();
  assertFalse(readOptions.useSignedStringMinMax());
  }