Github user Vlad-Storona commented on a diff in the pull request: https://github.com/apache/drill/pull/774#discussion_r148567854 --- Diff: contrib/storage-opentsdb/src/main/java/org/apache/drill/exec/store/openTSDB/OpenTSDBRecordReader.java --- @@ -0,0 +1,263 @@ +/* + * 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.drill.exec.store.openTSDB; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import org.apache.drill.common.exceptions.ExecutionSetupException; +import org.apache.drill.common.exceptions.UserException; +import org.apache.drill.common.expression.SchemaPath; +import org.apache.drill.common.logical.ValidationError; +import org.apache.drill.common.types.TypeProtos; +import org.apache.drill.common.types.TypeProtos.MajorType; +import org.apache.drill.common.types.TypeProtos.MinorType; +import org.apache.drill.common.types.Types; +import org.apache.drill.exec.exception.SchemaChangeException; +import org.apache.drill.exec.expr.TypeHelper; +import org.apache.drill.exec.ops.OperatorContext; +import org.apache.drill.exec.physical.impl.OutputMutator; +import org.apache.drill.exec.record.MaterializedField; +import org.apache.drill.exec.store.AbstractRecordReader; +import org.apache.drill.exec.store.openTSDB.client.OpenTSDBTypes; +import org.apache.drill.exec.store.openTSDB.client.Schema; +import org.apache.drill.exec.store.openTSDB.client.Service; +import org.apache.drill.exec.store.openTSDB.dto.ColumnDTO; +import org.apache.drill.exec.store.openTSDB.dto.MetricDTO; +import org.apache.drill.exec.vector.NullableFloat8Vector; +import org.apache.drill.exec.vector.NullableTimeStampVector; +import org.apache.drill.exec.vector.NullableVarCharVector; +import org.apache.drill.exec.vector.ValueVector; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static java.nio.charset.StandardCharsets.UTF_8; + +public class OpenTSDBRecordReader extends AbstractRecordReader { + + private static final Logger log = LoggerFactory.getLogger(OpenTSDBRecordReader.class); + + private static final Map<OpenTSDBTypes, MinorType> TYPES; + + private Service db; + + private Iterator<MetricDTO> tableIterator; + private OutputMutator output; + private ImmutableList<ProjectedColumnInfo> projectedCols; + private OpenTSDBSubScan.OpenTSDBSubScanSpec subScanSpec; + + OpenTSDBRecordReader(Service client, OpenTSDBSubScan.OpenTSDBSubScanSpec subScanSpec, + List<SchemaPath> projectedColumns) throws IOException { + setColumns(projectedColumns); + this.db = client; + this.subScanSpec = subScanSpec; + db.setupQueryParameters(subScanSpec.getTableName()); + log.debug("Scan spec: {}", subScanSpec); + } + + @Override + public void setup(OperatorContext context, OutputMutator output) throws ExecutionSetupException { + this.output = output; + Set<MetricDTO> tables = db.getTablesFromDB(); + if (tables == null || tables.isEmpty()) { + throw new ValidationError(String.format("Table '%s' not found or it's empty", subScanSpec.getTableName())); + } + this.tableIterator = tables.iterator(); + } + + @Override + public int next() { + try { + return processOpenTSDBTablesData(); + } catch (SchemaChangeException e) { + log.info(e.toString()); + return 0; + } + } + + @Override + protected boolean isSkipQuery() { --- End diff -- I will remove it.
---