forked from apache/drill
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
DRILL-7734: Revise the result set reader
Revised into two forms: push (for streaming JSON results) and pull (for one operator reading from another). closes apache#2077
- Loading branch information
1 parent
64b40be
commit 953280b
Showing
21 changed files
with
1,060 additions
and
605 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
49 changes: 49 additions & 0 deletions
49
...java-exec/src/main/java/org/apache/drill/exec/physical/resultSet/PushResultSetReader.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,49 @@ | ||
| /* | ||
| * 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.physical.resultSet; | ||
|
|
||
| import org.apache.drill.exec.physical.rowSet.RowSetReader; | ||
|
|
||
| /** | ||
| * Push-based result set reader, in which the caller obtains batches | ||
| * and registers them with the implementation. The client thus is responsible | ||
| * for detecting the end of batches and releasing memory. General protocol: | ||
| * <p> | ||
| * <ul> | ||
| * <li>Create an instance and bind it to a batch source.</li> | ||
| * <li>Obtain a batch, typically by having it passed in.</li> | ||
| * <li>Call {@link #start()} to obtain a reader for that batch.</li> | ||
| * <li>Iterate over the rows.</li> | ||
| * <li>Release memory for the batch.</li> | ||
| * </ul> | ||
| * <p> | ||
| * In Drill, | ||
| * batches may have the same or different schemas. Each call to | ||
| * {@link #start()} prepares a {@link RowSetReader} to use for | ||
| * the available batch. If the batch has the same schema as the previous, | ||
| * then the existing reader is simply repositioned at the start of the | ||
| * batch. If the schema changed (or this is the first batch), then a | ||
| * new reader is created. Thus, the client should not assume that the | ||
| * same reader is available across calls. However, if it is useful to | ||
| * cache column writers, simply check if the reader returned from | ||
| * {@code start()} is the same as the previous one. If so, the column | ||
| * writers are also the same. | ||
| */ | ||
| public interface PushResultSetReader { | ||
| RowSetReader start(); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
134 changes: 134 additions & 0 deletions
134
.../src/main/java/org/apache/drill/exec/physical/resultSet/impl/PullResultSetReaderImpl.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,134 @@ | ||
| /* | ||
| * 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.physical.resultSet.impl; | ||
|
|
||
| import org.apache.drill.exec.physical.impl.protocol.BatchAccessor; | ||
| import org.apache.drill.exec.physical.resultSet.PullResultSetReader; | ||
| import org.apache.drill.exec.physical.rowSet.RowSetReader; | ||
| import org.apache.drill.exec.record.metadata.TupleMetadata; | ||
| import org.apache.drill.shaded.guava.com.google.common.annotations.VisibleForTesting; | ||
| import org.apache.drill.shaded.guava.com.google.common.base.Preconditions; | ||
|
|
||
| /** | ||
| * <h4>Protocol</h4> | ||
| * <ol> | ||
| * <li>Create an instance, passing in a | ||
| * {@link UpstreamSource} to provide batches and optional | ||
| * selection vector.</li> | ||
| * <li>For each incoming batch: | ||
| * <ol> | ||
| * <li>Call {@link #start()} to attach the batch. The associated | ||
| * {@link BatchAccessor} reports if the schema has changed.</li> | ||
| * <li>Call {@link #reader()} to obtain a reader.</li> | ||
| * <li>Iterate over the batch using the reader.</li> | ||
| * <li>Call {@link #release()} to free the memory for the | ||
| * incoming batch. Or, to call {@link #detach()} to keep | ||
| * the batch memory.</li> | ||
| * </ol> | ||
| * <li>Call {@link #close()} after all batches are read.</li> | ||
| * </ol> | ||
| */ | ||
| public class PullResultSetReaderImpl implements PullResultSetReader { | ||
|
|
||
| public interface UpstreamSource extends PushResultSetReaderImpl.UpstreamSource { | ||
| boolean next(); | ||
| void release(); | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
| protected enum State { | ||
| START, | ||
| PENDING, | ||
| BATCH, | ||
| DETACHED, | ||
| EOF, | ||
| CLOSED | ||
| } | ||
|
|
||
| private final PushResultSetReaderImpl baseReader; | ||
| private final UpstreamSource source; | ||
| private State state = State.START; | ||
| private RowSetReader rowSetReader; | ||
|
|
||
| public PullResultSetReaderImpl(UpstreamSource source) { | ||
| this.baseReader = new PushResultSetReaderImpl(source); | ||
| this.source = source; | ||
| } | ||
|
|
||
| @Override | ||
| public TupleMetadata schema() { | ||
| switch (state) { | ||
| case CLOSED: | ||
| return null; | ||
| case START: | ||
| if (!next()) { | ||
| return null; | ||
| } | ||
| state = State.PENDING; | ||
| break; | ||
| default: | ||
| } | ||
| return rowSetReader.tupleSchema(); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean next() { | ||
| switch (state) { | ||
| case PENDING: | ||
| state = State.BATCH; | ||
| return true; | ||
| case BATCH: | ||
| source.release(); | ||
| break; | ||
| case CLOSED: | ||
| throw new IllegalStateException("Reader is closed"); | ||
| case EOF: | ||
| return false; | ||
| case START: | ||
| break; | ||
| default: | ||
| source.release(); | ||
| } | ||
| if (!source.next()) { | ||
| state = State.EOF; | ||
| return false; | ||
| } | ||
|
|
||
| rowSetReader = baseReader.start(); | ||
| state = State.BATCH; | ||
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| public int schemaVersion() { return source.schemaVersion(); } | ||
|
|
||
| @Override | ||
| public RowSetReader reader() { | ||
| Preconditions.checkState(state == State.BATCH, "Not in batch-ready state."); | ||
| return rowSetReader; | ||
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
| source.release(); | ||
| state = State.CLOSED; | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
| protected State state() { return state; } | ||
| } |
Oops, something went wrong.