You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tajo.apache.org by hy...@apache.org on 2013/09/04 17:51:13 UTC
[1/3] TAJO-144: Implement INSERT OVERWRITE clause. (hyunsik)
Updated Branches:
refs/heads/master b2dc84fe7 -> ffdb1c71c
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/worker/TaskRunnerTest.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/worker/TaskRunnerTest.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/worker/TaskRunnerTest.java
deleted file mode 100644
index 952cb0f..0000000
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/worker/TaskRunnerTest.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/**
- * 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.tajo.worker;
-
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
-import org.apache.hadoop.yarn.api.records.ContainerId;
-import org.apache.hadoop.yarn.api.records.NodeId;
-import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
-import org.apache.hadoop.yarn.util.BuilderUtils;
-import org.apache.tajo.*;
-import org.apache.tajo.ipc.TajoWorkerProtocol.TajoWorkerProtocolService;
-import org.apache.tajo.rpc.ProtoAsyncRpcClient;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-public class TaskRunnerTest {
- long ts1 = 1315890136000l;
- QueryId q1 = TestTajoIds.createQueryId(ts1, 2);
- ExecutionBlockId sq1 = QueryIdFactory.newExecutionBlockId(q1, 5);
-
- //@Test
- public void testInit() throws Exception {
- ProtoAsyncRpcClient mockClient = mock(ProtoAsyncRpcClient.class);
- mockClient.close();
-
- TajoWorkerProtocolService.Interface mockMaster =
- mock(TajoWorkerProtocolService.Interface.class);
- ApplicationAttemptId appAttemptId = BuilderUtils.newApplicationAttemptId(
- BuilderUtils.newApplicationId(Integer.parseInt(q1.getId()), q1.getSeq()), 1);
- ContainerId cId = BuilderUtils.newContainerId(appAttemptId, 1);
-
- NodeId nodeId = RecordFactoryProvider.getRecordFactory(null).
- newRecordInstance(NodeId.class);
- nodeId.setHost("host1");
- nodeId.setPort(9001);
- UserGroupInformation mockTaskOwner = mock(UserGroupInformation.class);
- when(mockTaskOwner.getShortUserName()).thenReturn("hyunsik");
- TaskRunner runner = new TaskRunner(sq1, nodeId, mockTaskOwner, mockMaster, cId);
- QueryConf conf = new QueryConf();
- conf.setOutputPath(new Path("/tmp/" + q1));
- runner.init(conf);
- runner.start();
- runner.stop();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/queries/insert_1.sql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/queries/insert_1.sql b/tajo-core/tajo-core-backend/src/test/queries/insert_1.sql
deleted file mode 100644
index e9f65c4..0000000
--- a/tajo-core/tajo-core-backend/src/test/queries/insert_1.sql
+++ /dev/null
@@ -1 +0,0 @@
-insert into people values (1, 'hyunsik', 32)
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/queries/insert_2.sql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/queries/insert_2.sql b/tajo-core/tajo-core-backend/src/test/queries/insert_2.sql
deleted file mode 100644
index 72d699b..0000000
--- a/tajo-core/tajo-core-backend/src/test/queries/insert_2.sql
+++ /dev/null
@@ -1 +0,0 @@
-insert into people (id, name, age) values (1, 'hyunsik', 32)
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/queries/insert_into_select_1.sql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/queries/insert_into_select_1.sql b/tajo-core/tajo-core-backend/src/test/queries/insert_into_select_1.sql
new file mode 100644
index 0000000..d3291f5
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/test/queries/insert_into_select_1.sql
@@ -0,0 +1 @@
+insert into table1 select col1, col2, sum(col3) from table2 group by col1, col2
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/queries/insert_into_select_2.sql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/queries/insert_into_select_2.sql b/tajo-core/tajo-core-backend/src/test/queries/insert_into_select_2.sql
new file mode 100644
index 0000000..2928b11
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/test/queries/insert_into_select_2.sql
@@ -0,0 +1 @@
+insert into location 'file:/tmp/data' select col1, col2, sum(col3) from table2 group by col1, col2
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/queries/insert_into_select_3.sql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/queries/insert_into_select_3.sql b/tajo-core/tajo-core-backend/src/test/queries/insert_into_select_3.sql
new file mode 100644
index 0000000..b7718dc
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/test/queries/insert_into_select_3.sql
@@ -0,0 +1 @@
+insert into table1 (col1, col2, col3) select col1, col2, sum(col3) as total from table2 group by col1, col2
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/queries/insert_overwrite_into_select_1.sql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/queries/insert_overwrite_into_select_1.sql b/tajo-core/tajo-core-backend/src/test/queries/insert_overwrite_into_select_1.sql
new file mode 100644
index 0000000..84992bc
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/test/queries/insert_overwrite_into_select_1.sql
@@ -0,0 +1 @@
+insert overwrite into table1 select col1, col2, sum(col3) from table2 group by col1, col2
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/queries/insert_overwrite_into_select_2.sql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/queries/insert_overwrite_into_select_2.sql b/tajo-core/tajo-core-backend/src/test/queries/insert_overwrite_into_select_2.sql
new file mode 100644
index 0000000..1c4227b
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/test/queries/insert_overwrite_into_select_2.sql
@@ -0,0 +1 @@
+insert overwrite into location 'file:/tmp/data' select col1, col2, sum(col3) from table2 group by col1, col2
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/queries/insert_overwrite_into_select_3.sql
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/queries/insert_overwrite_into_select_3.sql b/tajo-core/tajo-core-backend/src/test/queries/insert_overwrite_into_select_3.sql
new file mode 100644
index 0000000..bfacc01
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/test/queries/insert_overwrite_into_select_3.sql
@@ -0,0 +1 @@
+insert overwrite into table1 (col1, col2, col3) select col1, col2, sum(col3) as total from table2 group by col1, col2
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/CSVFile.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/CSVFile.java b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/CSVFile.java
index d8e6ea1..9231ead 100644
--- a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/CSVFile.java
+++ b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/CSVFile.java
@@ -32,9 +32,9 @@ import org.apache.tajo.catalog.TableMeta;
import org.apache.tajo.catalog.statistics.TableStat;
import org.apache.tajo.datum.*;
import org.apache.tajo.exception.UnsupportedException;
+import org.apache.tajo.storage.compress.CodecPool;
import org.apache.tajo.storage.exception.AlreadyExistsStorageException;
import org.apache.tajo.storage.json.StorageGsonHelper;
-import org.apache.tajo.storage.compress.CodecPool;
import java.io.*;
import java.util.Arrays;
@@ -503,6 +503,9 @@ public class CSVFile {
Datum.class);
tuple.put(tid, data);
break;
+ case NULL:
+ tuple.put(tid, NullDatum.get());
+ break;
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/RawFile.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/RawFile.java b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/RawFile.java
index 5c980d9..499b59b 100644
--- a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/RawFile.java
+++ b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/RawFile.java
@@ -28,6 +28,7 @@ import org.apache.tajo.common.TajoDataTypes.DataType;
import org.apache.tajo.datum.ArrayDatum;
import org.apache.tajo.datum.Datum;
import org.apache.tajo.datum.DatumFactory;
+import org.apache.tajo.datum.NullDatum;
import org.apache.tajo.json.CommonGsonHelper;
import org.apache.tajo.util.BitArray;
@@ -193,14 +194,6 @@ public class RawFile {
tuple.put(i, DatumFactory.createFloat8(buffer.getDouble()));
break;
-// case TEXT :
-// // TODO - shoud use CharsetEncoder / CharsetDecoder
-// int strSize = buffer.getInt();
-// byte [] strBytes = new byte[strSize];
-// buffer.get(strBytes);
-// tuple.put(i, DatumFactory.createText(new String(strBytes)));
-// break;
-
case TEXT :
// TODO - shoud use CharsetEncoder / CharsetDecoder
int strSize2 = buffer.getInt();
@@ -231,7 +224,11 @@ public class RawFile {
tuple.put(i, array);
break;
- default:
+ case NULL:
+ tuple.put(i, NullDatum.get());
+ break;
+
+ default:
}
}
@@ -379,6 +376,10 @@ public class RawFile {
}
switch(columnTypes[i].getType()) {
+ case NULL:
+ nullFlags.set(i);
+ continue;
+
case BOOLEAN:
case BIT:
buffer.put(t.get(i).asByte());
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/RowFile.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/RowFile.java b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/RowFile.java
index 0911102..b76d3c3 100644
--- a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/RowFile.java
+++ b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/RowFile.java
@@ -453,6 +453,9 @@ public class RowFile {
buffer.put(byteArray);
break;
}
+ case NULL:
+ nullFlags.set(i);
+ break;
default:
break;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/StorageManager.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/StorageManager.java b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/StorageManager.java
index e7479fd..db5b5d8 100644
--- a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/StorageManager.java
+++ b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/StorageManager.java
@@ -117,6 +117,25 @@ public class StorageManager {
fs.delete(tablePath, true);
}
+ public boolean exists(Path path) throws IOException {
+ FileSystem fileSystem = path.getFileSystem(conf);
+ return fileSystem.exists(path);
+ }
+
+ /**
+ * This method deletes only data contained in the given path.
+ *
+ * @param path The path in which data are deleted.
+ * @throws IOException
+ */
+ public void deleteData(Path path) throws IOException {
+ FileSystem fileSystem = path.getFileSystem(conf);
+ FileStatus[] fileLists = fileSystem.listStatus(path);
+ for (FileStatus status : fileLists) {
+ fileSystem.delete(status.getPath(), true);
+ }
+ }
+
public Path getTablePath(String tableName) {
return new Path(tableBaseDir, tableName);
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/rcfile/RCFileWrapper.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/rcfile/RCFileWrapper.java b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/rcfile/RCFileWrapper.java
index 6332416..251587e 100644
--- a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/rcfile/RCFileWrapper.java
+++ b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/rcfile/RCFileWrapper.java
@@ -123,6 +123,11 @@ public class RCFileWrapper {
byteRef.set(i, cu);
break;
}
+ case NULL:
+ cu = new BytesRefWritable(new byte[0]);
+ byteRef.set(i, cu);
+ break;
+
default:
throw new IOException("ERROR: Unsupported Data Type");
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/trevni/TrevniAppender.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/trevni/TrevniAppender.java b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/trevni/TrevniAppender.java
index eef95e0..f83382e 100644
--- a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/trevni/TrevniAppender.java
+++ b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/trevni/TrevniAppender.java
@@ -23,10 +23,6 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.trevni.ColumnFileMetaData;
-import org.apache.trevni.ColumnFileWriter;
-import org.apache.trevni.ColumnMetaData;
-import org.apache.trevni.ValueType;
import org.apache.tajo.catalog.Column;
import org.apache.tajo.catalog.TableMeta;
import org.apache.tajo.catalog.statistics.TableStat;
@@ -34,6 +30,10 @@ import org.apache.tajo.common.TajoDataTypes.Type;
import org.apache.tajo.storage.FileAppender;
import org.apache.tajo.storage.TableStatistics;
import org.apache.tajo.storage.Tuple;
+import org.apache.trevni.ColumnFileMetaData;
+import org.apache.trevni.ColumnFileWriter;
+import org.apache.trevni.ColumnMetaData;
+import org.apache.trevni.ValueType;
import java.io.FileNotFoundException;
import java.io.IOException;
@@ -110,6 +110,8 @@ public class TrevniAppender extends FileAppender {
return ValueType.BYTES;
case ARRAY:
return ValueType.BYTES;
+ case NULL:
+ return ValueType.NULL;
default:
return null;
}
@@ -132,6 +134,8 @@ public class TrevniAppender extends FileAppender {
if (!t.isNull(i)) {
col = schema.getColumn(i);
switch (col.getDataType().getType()) {
+ case NULL:
+ break;
case BOOLEAN:
case BIT:
case INT2:
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/trevni/TrevniScanner.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/trevni/TrevniScanner.java b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/trevni/TrevniScanner.java
index be23d48..0f164b0 100644
--- a/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/trevni/TrevniScanner.java
+++ b/tajo-core/tajo-core-storage/src/main/java/org/apache/tajo/storage/trevni/TrevniScanner.java
@@ -19,17 +19,18 @@
package org.apache.tajo.storage.trevni;
import org.apache.hadoop.conf.Configuration;
-import org.apache.trevni.ColumnFileReader;
-import org.apache.trevni.ColumnValues;
-import org.apache.trevni.avro.HadoopInput;
import org.apache.tajo.catalog.Column;
import org.apache.tajo.catalog.TableMeta;
import org.apache.tajo.datum.BlobDatum;
import org.apache.tajo.datum.DatumFactory;
+import org.apache.tajo.datum.NullDatum;
import org.apache.tajo.storage.FileScanner;
import org.apache.tajo.storage.Fragment;
import org.apache.tajo.storage.Tuple;
import org.apache.tajo.storage.VTuple;
+import org.apache.trevni.ColumnFileReader;
+import org.apache.trevni.ColumnValues;
+import org.apache.trevni.avro.HadoopInput;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -127,11 +128,6 @@ public class TrevniScanner extends FileScanner {
DatumFactory.createInet4(((ByteBuffer) columns[i].nextValue()).array()));
break;
-// case TEXT:
-// tuple.put(tid,
-// DatumFactory.createText((String) columns[i].nextValue()));
-// break;
-
case TEXT:
tuple.put(tid,
DatumFactory.createText((String) columns[i].nextValue()));
@@ -142,6 +138,10 @@ public class TrevniScanner extends FileScanner {
new BlobDatum(((ByteBuffer) columns[i].nextValue())));
break;
+ case NULL:
+ tuple.put(tid, NullDatum.get());
+ break;
+
default:
throw new IOException("Unsupport data type");
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-storage/src/test/java/org/apache/tajo/storage/TestStorages.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-storage/src/test/java/org/apache/tajo/storage/TestStorages.java b/tajo-core/tajo-core-storage/src/test/java/org/apache/tajo/storage/TestStorages.java
index 22207fe..bde9835 100644
--- a/tajo-core/tajo-core-storage/src/test/java/org/apache/tajo/storage/TestStorages.java
+++ b/tajo-core/tajo-core-storage/src/test/java/org/apache/tajo/storage/TestStorages.java
@@ -21,9 +21,6 @@ package org.apache.tajo.storage;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
import org.apache.tajo.catalog.CatalogUtil;
import org.apache.tajo.catalog.Options;
import org.apache.tajo.catalog.Schema;
@@ -34,8 +31,12 @@ import org.apache.tajo.common.TajoDataTypes.Type;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.datum.Datum;
import org.apache.tajo.datum.DatumFactory;
+import org.apache.tajo.datum.NullDatum;
import org.apache.tajo.storage.rcfile.RCFile;
import org.apache.tajo.util.CommonTestingUtil;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
import java.io.IOException;
import java.util.Arrays;
@@ -194,6 +195,7 @@ public class TestStorages {
schema.addColumn("col9", Type.TEXT);
schema.addColumn("col10", Type.BLOB);
schema.addColumn("col11", Type.INET4);
+ schema.addColumn("col12", Type.NULL);
Options options = new Options();
TableMeta meta = CatalogUtil.newTableMeta(schema, storeType, options);
@@ -202,7 +204,7 @@ public class TestStorages {
Appender appender = StorageManager.getAppender(conf, meta, tablePath);
appender.init();
- Tuple tuple = new VTuple(11);
+ Tuple tuple = new VTuple(12);
tuple.put(new Datum[] {
DatumFactory.createBool(true),
DatumFactory.createBit((byte) 0x99),
@@ -215,7 +217,7 @@ public class TestStorages {
DatumFactory.createText("hyunsik"),
DatumFactory.createBlob("hyunsik".getBytes()),
DatumFactory.createInet4("192.168.0.1"),
- DatumFactory.createText("hyunsik")
+ NullDatum.get()
});
appender.addTuple(tuple);
appender.flush();
[2/3] TAJO-144: Implement INSERT OVERWRITE clause. (hyunsik)
Posted by hy...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/QueryMeta.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/QueryMeta.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/QueryMeta.java
new file mode 100644
index 0000000..6655317
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/QueryMeta.java
@@ -0,0 +1,118 @@
+/**
+ * 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.tajo.master;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.catalog.Options;
+import org.apache.tajo.conf.TajoConf;
+import org.apache.tajo.engine.planner.InsertNode;
+import org.apache.tajo.engine.planner.PlannerUtil;
+import org.apache.tajo.engine.planner.logical.CreateTableNode;
+
+import static org.apache.tajo.catalog.proto.CatalogProtos.KeyValueSetProto;
+
+public class QueryMeta extends Options {
+
+ public QueryMeta() {}
+
+ public QueryMeta(KeyValueSetProto proto) {
+ super(proto);
+ }
+
+ public void put(TajoConf.ConfVars key, String value) {
+ put(key.varname, value);
+ }
+
+ public String get(TajoConf.ConfVars key) {
+ return get(key.varname);
+ }
+
+ public String get(String key) {
+ return super.get(key);
+ }
+
+ public void setUser(String username) {
+ put(TajoConf.ConfVars.QUERY_USERNAME, username);
+ }
+
+ public String getUser() {
+ return get(TajoConf.ConfVars.QUERY_USERNAME);
+ }
+
+ public void setOutputTable(String tableName) {
+ put(TajoConf.ConfVars.QUERY_OUTPUT_TABLE, PlannerUtil.normalizeTableName(tableName));
+ }
+
+ public String getOutputTable() {
+ return get(TajoConf.ConfVars.QUERY_OUTPUT_TABLE);
+ }
+
+ public void setOutputPath(Path path) {
+ // it is determined in QueryMaster.initStagingDir().
+ put(TajoConf.ConfVars.QUERY_OUTPUT_DIR, path.toUri().toString());
+ }
+
+ public Path getOutputPath() {
+ return new Path(get(TajoConf.ConfVars.QUERY_OUTPUT_DIR));
+ }
+
+ public void setOutputOverwrite() {
+ put("tajo.output.overwrite", "1");
+ }
+
+ public boolean isOutputOverwrite() {
+ String overwrite = get("tajo.output.overwrite");
+ if (overwrite != null && overwrite.equals("1")) {
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ public void setFileOutput() {
+ put("tajo.output.fileoutput", "1");
+ }
+
+ public boolean isFileOutput() {
+ String fileoutput = get("tajo.output.fileoutput");
+ if (fileoutput != null && fileoutput.equals("1")) {
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ public void setCreateTable() {
+ put("tajo.query.command", CreateTableNode.class.getSimpleName());
+ }
+
+ public boolean isCreateTable() {
+ String command = get("tajo.query.command");
+ return command != null && command.equals(CreateTableNode.class.getSimpleName());
+ }
+
+ public void setInsert() {
+ put("tajo.query.command", InsertNode.class.getSimpleName());
+ }
+
+ public boolean isInsert() {
+ String command = get("tajo.query.command");
+ return command != null && command.equals(InsertNode.class.getSimpleName());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TaskSchedulerImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TaskSchedulerImpl.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TaskSchedulerImpl.java
index 9b454f6..b26d05b 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TaskSchedulerImpl.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/TaskSchedulerImpl.java
@@ -118,7 +118,6 @@ public class TaskSchedulerImpl extends AbstractService
schedule();
}
- //req.getCallback().run(stopTaskRunnerReq);
LOG.info("TaskScheduler schedulingThread stopped");
}
};
@@ -459,7 +458,8 @@ public class TaskSchedulerImpl extends AbstractService
new ArrayList<Fragment>(task.getAllFragments()),
task.getOutputName(),
false,
- task.getLogicalPlan().toJson());
+ task.getLogicalPlan().toJson(),
+ context.getQueryMeta());
if (task.getStoreTableNode().isLocal()) {
taskAssign.setInterQuery();
}
@@ -503,7 +503,8 @@ public class TaskSchedulerImpl extends AbstractService
Lists.newArrayList(task.getAllFragments()),
task.getOutputName(),
false,
- task.getLogicalPlan().toJson());
+ task.getLogicalPlan().toJson(),
+ context.getQueryMeta());
if (task.getStoreTableNode().isLocal()) {
taskAssign.setInterQuery();
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/event/QueryStartEvent.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/event/QueryStartEvent.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/event/QueryStartEvent.java
index 6ae8ff7..54cb0fa 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/event/QueryStartEvent.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/event/QueryStartEvent.java
@@ -20,18 +20,24 @@ package org.apache.tajo.master.event;
import org.apache.hadoop.yarn.event.AbstractEvent;
import org.apache.tajo.QueryId;
+import org.apache.tajo.master.QueryMeta;
+/**
+ * This event is conveyed to QueryMaster.
+ */
public class QueryStartEvent extends AbstractEvent {
public enum EventType {
QUERY_START
}
private QueryId queryId;
+ private QueryMeta queryMeta;
private String logicalPlanJson;
- public QueryStartEvent(QueryId queryId, String logicalPlanJson) {
+ public QueryStartEvent(QueryId queryId, QueryMeta queryMeta, String logicalPlanJson) {
super(EventType.QUERY_START);
this.queryId = queryId;
+ this.queryMeta = queryMeta;
this.logicalPlanJson = logicalPlanJson;
}
@@ -39,6 +45,10 @@ public class QueryStartEvent extends AbstractEvent {
return queryId;
}
+ public QueryMeta getQueryMeta() {
+ return this.queryMeta;
+ }
+
public String getLogicalPlanJson() {
return logicalPlanJson;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Query.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Query.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Query.java
index 99b7c62..d1e92be 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Query.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/Query.java
@@ -29,9 +29,13 @@ import org.apache.tajo.ExecutionBlockId;
import org.apache.tajo.QueryConf;
import org.apache.tajo.QueryId;
import org.apache.tajo.TajoProtos.QueryState;
+import org.apache.tajo.catalog.CatalogService;
import org.apache.tajo.catalog.TableDesc;
import org.apache.tajo.catalog.TableDescImpl;
+import org.apache.tajo.engine.planner.PlannerUtil;
import org.apache.tajo.engine.planner.global.MasterPlan;
+import org.apache.tajo.engine.planner.logical.NodeType;
+import org.apache.tajo.engine.planner.logical.StoreTableNode;
import org.apache.tajo.master.ExecutionBlock;
import org.apache.tajo.master.ExecutionBlockCursor;
import org.apache.tajo.master.event.*;
@@ -57,7 +61,7 @@ public class Query implements EventHandler<QueryEvent> {
private final EventHandler eventHandler;
private final MasterPlan plan;
private final StorageManager sm;
- private QueryMasterTask.QueryContext context;
+ QueryMasterTask.QueryContext context;
private ExecutionBlockCursor cursor;
// Query Status
@@ -305,19 +309,29 @@ public class Query implements EventHandler<QueryEvent> {
} else { // Finish a query
if (query.checkQueryForCompleted() == QueryState.QUERY_SUCCEEDED) {
SubQuery subQuery = query.getSubQuery(castEvent.getExecutionBlockId());
- TableDesc desc = new TableDescImpl(query.conf.getOutputTable(),
- subQuery.getTableMeta(), query.context.getOutputPath());
- query.setResultDesc(desc);
- try {
- query.writeStat(query.context.getOutputPath(), subQuery);
- } catch (IOException e) {
- e.printStackTrace();
+ TableDesc outputTableDesc = new TableDescImpl(query.context.getQueryMeta().getOutputTable(),
+ subQuery.getTableMeta(), query.context.getQueryMeta().getOutputPath());
+ query.setResultDesc(outputTableDesc);
+
+ if (!query.context.getQueryMeta().isFileOutput()) {
+ try {
+ query.writeStat(query.context.getQueryMeta().getOutputPath(), subQuery);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
}
query.eventHandler.handle(new QueryFinishEvent(query.getId()));
- if (query.context.isCreateTableQuery()) {
- // TOOD move to QueryJobManager
- //query.context.getCatalog().addTable(desc);
+ StoreTableNode storeTableNode = (StoreTableNode) PlannerUtil.findTopNode(subQuery.getBlock().getPlan(),
+ NodeType.STORE);
+ if (storeTableNode.isCreatedTable()) {
+ query.context.getQueryMasterContext().getWorkerContext().getCatalog().addTable(outputTableDesc);
+ } else if (storeTableNode.isOverwrite() && !query.context.getQueryMeta().isFileOutput()) {
+ CatalogService catalog = query.context.getQueryMasterContext().getWorkerContext().getCatalog();
+ TableDesc updatingTable = catalog.getTableDesc(outputTableDesc.getName());
+ updatingTable.getMeta().setStat(outputTableDesc.getMeta().getStat());
+ catalog.deleteTable(outputTableDesc.getName());
+ catalog.addTable(updatingTable);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
index 10be0a3..f6cf02e 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryInProgress.java
@@ -29,6 +29,7 @@ import org.apache.tajo.QueryId;
import org.apache.tajo.TajoProtos;
import org.apache.tajo.engine.planner.logical.LogicalRootNode;
import org.apache.tajo.ipc.TajoWorkerProtocol;
+import org.apache.tajo.master.QueryMeta;
import org.apache.tajo.master.TajoAsyncDispatcher;
import org.apache.tajo.master.TajoMaster;
import org.apache.tajo.master.rm.WorkerResource;
@@ -45,6 +46,8 @@ public class QueryInProgress extends CompositeService {
private QueryId queryId;
+ private QueryMeta queryMeta;
+
private TajoAsyncDispatcher dispatcher;
private LogicalRootNode plan;
@@ -63,9 +66,11 @@ public class QueryInProgress extends CompositeService {
public QueryInProgress(
TajoMaster.MasterContext masterContext,
+ QueryMeta queryMeta,
QueryId queryId, String sql, LogicalRootNode plan) {
super(QueryInProgress.class.getName());
this.masterContext = masterContext;
+ this.queryMeta = queryMeta;
this.queryId = queryId;
this.plan = plan;
@@ -195,12 +200,13 @@ public class QueryInProgress extends CompositeService {
//TODO wait
return;
}
- LOG.info("====>Call executeQuery to :" +
+ LOG.info("Call executeQuery to :" +
queryInfo.getQueryMasterHost() + ":" + queryInfo.getQueryMasterPort() + "," + queryId);
queryMasterRpcClient.executeQuery(
null,
TajoWorkerProtocol.QueryExecutionRequestProto.newBuilder()
.setQueryId(queryId.getProto())
+ .setQueryMeta(queryMeta.getProto())
.setLogicalPlanJson(PrimitiveProtos.StringProto.newBuilder().setValue(plan.toJson()).build())
.build(), NullCallback.get());
querySubmitted.set(true);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryJobManager.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryJobManager.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryJobManager.java
index 16008fc..fb92616 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryJobManager.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryJobManager.java
@@ -28,6 +28,7 @@ import org.apache.tajo.QueryId;
import org.apache.tajo.QueryIdFactory;
import org.apache.tajo.engine.planner.logical.LogicalRootNode;
import org.apache.tajo.ipc.TajoMasterProtocol;
+import org.apache.tajo.master.QueryMeta;
import org.apache.tajo.master.TajoMaster;
import org.apache.tajo.master.rm.WorkerResource;
@@ -84,9 +85,9 @@ public class QueryJobManager extends CompositeService {
return dispatcher.getEventHandler();
}
- public QueryInfo createNewQueryJob(String sql, LogicalRootNode plan) throws Exception {
+ public QueryInfo createNewQueryJob(QueryMeta queryMeta, String sql, LogicalRootNode plan) throws Exception {
QueryId queryId = QueryIdFactory.newQueryId(masterContext.getResourceManager().getSeedQueryId());
- QueryInProgress queryInProgress = new QueryInProgress(masterContext, queryId, sql, plan);
+ QueryInProgress queryInProgress = new QueryInProgress(masterContext,queryMeta, queryId, sql, plan);
synchronized(runningQueries) {
runningQueries.put(queryId, queryInProgress);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
index 50ec5be..53b9c05 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMaster.java
@@ -60,7 +60,6 @@ public class QueryMaster extends CompositeService implements EventHandler {
private GlobalOptimizer globalOptimizer;
-// private boolean isCreateTableStmt;
private StorageManager storageManager;
private QueryConf queryConf;
@@ -88,7 +87,7 @@ public class QueryMaster extends CompositeService implements EventHandler {
queryConf = new QueryConf(conf);
queryConf.addResource(new Path(QueryConf.QUERY_MASTER_FILENAME));
- QUERY_SESSION_TIMEOUT = 60 * 1000;//queryConf.getIntVar(TajoConf.ConfVars.QUERY_SESSION_TIMEOUT);
+ QUERY_SESSION_TIMEOUT = 60 * 1000;
queryMasterContext = new QueryMasterContext(queryConf);
clock = new SystemClock();
@@ -112,7 +111,7 @@ public class QueryMaster extends CompositeService implements EventHandler {
@Override
public void start() {
- LOG.info("====>QueryMaster start");
+ LOG.info("QueryMaster start");
queryHeartbeatThread = new QueryHeartbeatThread();
queryHeartbeatThread.start();
@@ -253,10 +252,10 @@ public class QueryMaster extends CompositeService implements EventHandler {
private class QueryStartEventHandler implements EventHandler<QueryStartEvent> {
@Override
public void handle(QueryStartEvent event) {
- LOG.info("====>Start QueryStartEventHandler:" + event.getQueryId());
+ LOG.info("Start QueryStartEventHandler:" + event.getQueryId());
//To change body of implemented methods use File | Settings | File Templates.
QueryMasterTask queryMasterTask = new QueryMasterTask(queryMasterContext,
- event.getQueryId(), event.getLogicalPlanJson());
+ event.getQueryId(), event.getQueryMeta(), event.getLogicalPlanJson());
queryMasterTask.init(queryConf);
queryMasterTask.start();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
index cfea8a2..f348e44 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/QueryMasterTask.java
@@ -39,6 +39,7 @@ import org.apache.tajo.engine.planner.logical.LogicalNode;
import org.apache.tajo.engine.planner.logical.LogicalRootNode;
import org.apache.tajo.engine.planner.logical.NodeType;
import org.apache.tajo.engine.planner.logical.ScanNode;
+import org.apache.tajo.master.QueryMeta;
import org.apache.tajo.master.TajoAsyncDispatcher;
import org.apache.tajo.master.event.*;
import org.apache.tajo.master.rm.TajoWorkerResourceManager;
@@ -64,6 +65,8 @@ public class QueryMasterTask extends CompositeService {
private QueryId queryId;
+ private QueryMeta queryMeta;
+
private QueryContext queryContext;
private QueryMaster.QueryMasterContext queryMasterContext;
@@ -78,8 +81,6 @@ public class QueryMasterTask extends CompositeService {
private final long querySubmitTime;
- private boolean isCreateTableStmt;
-
private Path outputPath;
private Map<String, TableDesc> tableDescMap = new HashMap<String, TableDesc>();
@@ -93,10 +94,11 @@ public class QueryMasterTask extends CompositeService {
private AtomicBoolean stopped = new AtomicBoolean(false);
public QueryMasterTask(QueryMaster.QueryMasterContext queryMasterContext,
- QueryId queryId, String logicalPlanJson) {
+ QueryId queryId, QueryMeta queryMeta, String logicalPlanJson) {
super(QueryMasterTask.class.getName());
this.queryMasterContext = queryMasterContext;
this.queryId = queryId;
+ this.queryMeta = queryMeta;
this.logicalPlanJson = logicalPlanJson;
this.querySubmitTime = System.currentTimeMillis();
}
@@ -269,13 +271,12 @@ public class QueryMasterTask extends CompositeService {
realUser = ugi.getShortUserName();
currentUser = UserGroupInformation.getCurrentUser().getShortUserName();
- String givenOutputTableName = queryConf.getOutputTable();
+ String givenOutputTableName = queryMeta.getOutputTable();
Path stagingDir;
// If final output directory is not given by an user,
// we use the query id as a output directory.
if (givenOutputTableName == null || givenOutputTableName.isEmpty()) {
- this.isCreateTableStmt = false;
FileSystem defaultFS = FileSystem.get(queryConf);
Path homeDirectory = defaultFS.getHomeDirectory();
@@ -318,25 +319,31 @@ public class QueryMasterTask extends CompositeService {
}
// Set the query id to the output table name
- queryConf.setOutputTable(queryId.toString());
+ queryMeta.setOutputTable(queryId.toString());
+
+ } else { // if a output table is given
- } else {
- this.isCreateTableStmt = true;
Path warehouseDir = new Path(queryConf.getVar(TajoConf.ConfVars.ROOT_DIR),
TajoConstants.WAREHOUSE_DIR);
- stagingDir = new Path(warehouseDir, queryConf.getOutputTable());
-
FileSystem fs = warehouseDir.getFileSystem(queryConf);
- if (fs.exists(stagingDir)) {
- throw new IOException("The staging directory " + stagingDir
- + " already exists. The directory must be unique to each query");
+
+ if (queryMeta.isFileOutput()) {
+ stagingDir = queryMeta.getOutputPath();
} else {
- // TODO - should have appropriate permission
- fs.mkdirs(stagingDir, new FsPermission(USER_DIR_PERMISSION));
+ stagingDir = new Path(warehouseDir, queryMeta.getOutputTable());
}
+
+ if (!queryMeta.isOutputOverwrite()) {
+ if (fs.exists(stagingDir)) {
+ throw new IOException("The staging directory " + stagingDir
+ + " already exists. The directory must be unique to each query");
+ }
+ }
+
+ fs.mkdirs(stagingDir, new FsPermission(USER_DIR_PERMISSION));
}
- queryConf.setOutputPath(stagingDir);
+ queryMeta.setOutputPath(stagingDir);
outputPath = stagingDir;
LOG.info("Initialized Query Staging Dir: " + outputPath);
}
@@ -383,6 +390,10 @@ public class QueryMasterTask extends CompositeService {
return queryMasterContext;
}
+ public QueryMeta getQueryMeta() {
+ return queryMeta;
+ }
+
public QueryConf getConf() {
return queryConf;
}
@@ -407,10 +418,6 @@ public class QueryMasterTask extends CompositeService {
return outputPath;
}
- public boolean isCreateTableQuery() {
- return isCreateTableStmt;
- }
-
public synchronized EventHandler getEventHandler() {
if(eventHandler == null) {
eventHandler = dispatcher.getEventHandler();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
index 89d3fed..ecde59c 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/querymaster/SubQuery.java
@@ -46,11 +46,8 @@ import org.apache.tajo.engine.planner.logical.GroupbyNode;
import org.apache.tajo.engine.planner.logical.NodeType;
import org.apache.tajo.engine.planner.logical.ScanNode;
import org.apache.tajo.engine.planner.logical.StoreTableNode;
-import org.apache.tajo.master.ExecutionBlock;
-import org.apache.tajo.master.TaskRunnerGroupEvent;
+import org.apache.tajo.master.*;
import org.apache.tajo.master.TaskRunnerGroupEvent.EventType;
-import org.apache.tajo.master.TaskScheduler;
-import org.apache.tajo.master.TaskSchedulerImpl;
import org.apache.tajo.master.event.*;
import org.apache.tajo.storage.Fragment;
import org.apache.tajo.storage.StorageManager;
@@ -72,6 +69,7 @@ public class SubQuery implements EventHandler<SubQueryEvent> {
private static final Log LOG = LogFactory.getLog(SubQuery.class);
+ private QueryMeta queryMeta;
private ExecutionBlock block;
private int priority;
private TableMeta meta;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
index e6ec9c0..1e9d6e1 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoResourceAllocator.java
@@ -189,7 +189,7 @@ public class TajoResourceAllocator extends AbstractResourceAllocator {
queryConfPath = new Path(queryConfPath, QueryConf.FILENAME);
if(!fs.exists(queryConfPath)){
- LOG.info("Writing a QueryConf to HDFS and add to local environment, outputPath=" + queryConf.getOutputPath());
+ LOG.info("Writing a QueryConf to HDFS and add to local environment, outputPath=" + queryConfPath);
writeConf(queryConf, queryConfPath);
} else {
LOG.warn("QueryConf already exist. path: " + queryConfPath.toString());
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorker.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorker.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorker.java
index bb27ee4..0fc896b 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorker.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorker.java
@@ -21,13 +21,14 @@ package org.apache.tajo.worker;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.service.CompositeService;
import org.apache.hadoop.yarn.util.RackResolver;
import org.apache.tajo.QueryId;
import org.apache.tajo.TajoProtos;
+import org.apache.tajo.catalog.CatalogClient;
+import org.apache.tajo.catalog.CatalogService;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.ipc.TajoMasterProtocol;
import org.apache.tajo.master.querymaster.QueryMaster;
@@ -36,11 +37,13 @@ import org.apache.tajo.pullserver.TajoPullServerService;
import org.apache.tajo.rpc.CallFuture2;
import org.apache.tajo.rpc.ProtoAsyncRpcClient;
import org.apache.tajo.rpc.protocolrecords.PrimitiveProtos;
+import org.apache.tajo.util.NetUtils;
import org.apache.tajo.util.TajoIdUtils;
import org.apache.tajo.webapp.StaticHttpServer;
import java.io.BufferedReader;
import java.io.File;
+import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.util.ArrayList;
@@ -64,11 +67,15 @@ public class TajoWorker extends CompositeService {
private TajoWorkerManagerService tajoWorkerManagerService;
+ private InetSocketAddress tajoMasterAddress;
+
//to TajoMaster
private ProtoAsyncRpcClient tajoMasterRpc;
private TajoMasterProtocol.TajoMasterProtocolService tajoMasterRpcClient;
+ private CatalogClient catalogClient;
+
private WorkerContext workerContext;
private TaskRunnerManager taskRunnerManager;
@@ -171,17 +178,14 @@ public class TajoWorker extends CompositeService {
workerHeartbeatThread.interrupt();
}
-// try {
-// FileSystem.closeAll();
-// } catch (IOException e) {
-// LOG.error(e.getMessage(), e);
-// }
+ if (catalogClient != null) {
+ catalogClient.close();
+ }
+
if(tajoMasterRpc != null) {
tajoMasterRpc.close();
}
-// for(Service eachService: getServices()) {
-// System.out.println("Service:" + eachService);
-// }
+
super.stop();
LOG.info("TajoWorker main thread exiting");
}
@@ -207,6 +211,10 @@ public class TajoWorker extends CompositeService {
return taskRunnerManager;
}
+ public CatalogService getCatalog() {
+ return catalogClient;
+ }
+
public TajoPullServerService getPullService() {
return pullService;
}
@@ -246,9 +254,10 @@ public class TajoWorker extends CompositeService {
private void setWorkerMode(String[] params) {
if("qm".equals(daemonMode)) {
//QueryMaster mode
- String tajoMasterAddress = params[2];
+ String tajoMasterAddress = params[2];
connectToTajoMaster(tajoMasterAddress);
+ connectToCatalog();
QueryId queryId = TajoIdUtils.parseQueryId(params[1]);
tajoWorkerManagerService.getQueryMaster().reportQueryStatusToQueryMaster(
@@ -259,21 +268,24 @@ public class TajoWorker extends CompositeService {
} else {
//Standby mode
connectToTajoMaster(tajoConf.get("tajo.master.manager.addr"));
+ connectToCatalog();
workerHeartbeatThread = new WorkerHeartbeatThread();
workerHeartbeatThread.start();
}
}
- private void connectToTajoMaster(String tajoMasterAddress) {
- LOG.info("Init TajoMaster connection to:" + tajoMasterAddress);
- InetSocketAddress addr = NetUtils.createSocketAddr(tajoMasterAddress);
+ private void connectToTajoMaster(String tajoMasterAddrString) {
+ LOG.info("Connecting to TajoMaster (" + tajoMasterAddrString +")");
+ this.tajoMasterAddress = NetUtils.createSocketAddr(tajoMasterAddrString);
+
while(true) {
try {
- tajoMasterRpc = new ProtoAsyncRpcClient(TajoMasterProtocol.class, addr);
+ tajoMasterRpc = new ProtoAsyncRpcClient(TajoMasterProtocol.class, this.tajoMasterAddress);
tajoMasterRpcClient = tajoMasterRpc.getStub();
break;
} catch (Exception e) {
- LOG.error("Can't connect to TajoMaster[" + addr + "], " + e.getMessage(), e);
+ LOG.error("Can't connect to TajoMaster[" + NetUtils.normalizeInetSocketAddress(tajoMasterAddress) + "], "
+ + e.getMessage(), e);
}
try {
@@ -283,6 +295,17 @@ public class TajoWorker extends CompositeService {
}
}
+ private void connectToCatalog() {
+ // TODO: To be improved. it's a hack. It assumes that CatalogServer is embedded in TajoMaster.
+ String hostName = this.tajoMasterAddress.getHostName();
+ int port = Integer.parseInt(tajoConf.getVar(TajoConf.ConfVars.CATALOG_ADDRESS).split(":")[1]);
+ try {
+ catalogClient = new CatalogClient(hostName, port);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
class WorkerHeartbeatThread extends Thread {
TajoMasterProtocol.ServerStatusProto.System systemInfo;
List<TajoMasterProtocol.ServerStatusProto.Disk> diskInfos =
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
index 3969b36..e1fd88a 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerClientService.java
@@ -100,11 +100,11 @@ public class TajoWorkerClientService extends AbstractService {
@Override
public void stop() {
- LOG.info("====> TajoWorkerClientService stopping");
+ LOG.info("TajoWorkerClientService stopping");
if(rpcServer != null) {
rpcServer.shutdown();
}
- LOG.info("====> TajoWorkerClientService stopped");
+ LOG.info("TajoWorkerClientService stopped");
super.stop();
}
@@ -174,7 +174,10 @@ public class TajoWorkerClientService extends AbstractService {
builder.setProgress(query.getProgress());
builder.setSubmitTime(query.getAppSubmitTime());
builder.setInitTime(query.getInitializationTime());
- builder.setHasResult(!queryMasterTask.getQueryContext().isCreateTableQuery());
+ builder.setHasResult(
+ !(queryMasterTask.getQueryContext().getQueryMeta().isCreateTable() ||
+ queryMasterTask.getQueryContext().getQueryMeta().isInsert())
+ );
if (query.getState() == TajoProtos.QueryState.QUERY_SUCCEEDED) {
builder.setFinishTime(query.getFinishTime());
} else {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerManagerService.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerManagerService.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerManagerService.java
index a48339a..6fde6e4 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerManagerService.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TajoWorkerManagerService.java
@@ -31,6 +31,7 @@ import org.apache.tajo.QueryUnitAttemptId;
import org.apache.tajo.TajoIdProtos;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.ipc.TajoWorkerProtocol;
+import org.apache.tajo.master.QueryMeta;
import org.apache.tajo.master.TaskSchedulerImpl;
import org.apache.tajo.master.event.*;
import org.apache.tajo.master.querymaster.QueryMaster;
@@ -199,8 +200,9 @@ public class TajoWorkerManagerService extends CompositeService
RpcCallback<PrimitiveProtos.BoolProto> done) {
try {
QueryId queryId = new QueryId(request.getQueryId());
- LOG.info("====>Receive executeQuery request:" + queryId);
- queryMaster.handle(new QueryStartEvent(queryId, request.getLogicalPlanJson().getValue()));
+ LOG.info("Receive executeQuery request:" + queryId);
+ queryMaster.handle(new QueryStartEvent(queryId,
+ new QueryMeta(request.getQueryMeta()), request.getLogicalPlanJson().getValue()));
done.run(TajoWorker.TRUE_PROTO);
} catch (Exception e) {
LOG.error(e.getMessage(), e);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java
index f9cc82c..aef5ead 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/Task.java
@@ -48,6 +48,7 @@ import org.apache.tajo.ipc.TajoWorkerProtocol.*;
import org.apache.tajo.ipc.TajoWorkerProtocol.TajoWorkerProtocolService.Interface;
import org.apache.tajo.ipc.protocolrecords.QueryUnitRequest;
import org.apache.tajo.master.ExecutionBlock.PartitionType;
+import org.apache.tajo.master.QueryMeta;
import org.apache.tajo.rpc.NullCallback;
import org.apache.tajo.storage.Fragment;
import org.apache.tajo.storage.StorageUtil;
@@ -69,6 +70,7 @@ public class Task {
private static final Log LOG = LogFactory.getLog(Task.class);
private final QueryConf conf;
+ private final QueryMeta queryMeta;
private final FileSystem localFS;
private final TaskRunner.TaskRunnerContext taskRunnerContext;
private final Interface masterProxy;
@@ -136,6 +138,7 @@ public class Task {
this.taskId = request.getId();
this.conf = worker.getQueryConf();
+ this.queryMeta = request.getQueryMeta();
this.taskRunnerContext = worker;
this.masterProxy = masterProxy;
this.localFS = worker.getLocalFS();
@@ -160,7 +163,7 @@ public class Task {
} else {
// The final result of a task will be written in a file named part-ss-nnnnnnn,
// where ss is the subquery id associated with this task, and nnnnnn is the task id.
- Path outFilePath = new Path(conf.getOutputPath(),
+ Path outFilePath = new Path(queryMeta.getOutputPath(),
OUTPUT_FILE_PREFIX +
OUTPUT_FILE_FORMAT_SUBQUERY.get().format(taskId.getQueryUnitId().getExecutionBlockId().getId()) + "-" +
OUTPUT_FILE_FORMAT_TASK.get().format(taskId.getQueryUnitId().getId()));
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunner.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunner.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunner.java
index 8493be2..5802ade 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunner.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/worker/TaskRunner.java
@@ -39,7 +39,6 @@ import org.apache.tajo.TajoProtos.TaskAttemptState;
import org.apache.tajo.conf.TajoConf.ConfVars;
import org.apache.tajo.engine.query.QueryUnitRequestImpl;
import org.apache.tajo.ipc.TajoWorkerProtocol;
-import org.apache.tajo.ipc.TajoWorkerProtocol.TajoWorkerProtocolService.Interface;
import org.apache.tajo.rpc.CallFuture2;
import org.apache.tajo.rpc.NullCallback;
import org.apache.tajo.rpc.ProtoAsyncRpcClient;
@@ -106,20 +105,6 @@ public class TaskRunner extends AbstractService {
private TaskRunnerManager taskRunnerManager;
- public TaskRunner(
- final ExecutionBlockId executionBlockId,
- final NodeId nodeId,
- UserGroupInformation taskOwner,
- Interface master, ContainerId containerId) {
- super(TaskRunner.class.getName());
- this.executionBlockId = executionBlockId;
- this.queryId = executionBlockId.getQueryId();
- this.nodeId = nodeId;
- this.taskOwner = taskOwner;
- this.master = master;
- this.containerId = containerId;
- }
-
public TaskRunner(TaskRunnerManager taskRunnerManager, QueryConf conf, String[] args) {
super(TaskRunner.class.getName());
@@ -127,10 +112,7 @@ public class TaskRunner extends AbstractService {
try {
final ExecutionBlockId executionBlockId = TajoIdUtils.createExecutionBlockId(args[1]);
- conf.setOutputPath(new Path(args[6]));
-
LOG.info("NM Local Dir: " + conf.get(ConfVars.TASK_LOCAL_DIR.varname));
- LOG.info("OUTPUT DIR: " + conf.getOutputPath());
LOG.info("Tajo Root Dir: " + conf.getVar(ConfVars.ROOT_DIR));
UserGroupInformation.setConfiguration(conf);
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto b/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
index 88a2029..0694b4e 100644
--- a/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
+++ b/tajo-core/tajo-core-backend/src/main/proto/TajoWorkerProtocol.proto
@@ -59,6 +59,7 @@ message QueryUnitRequestProto {
optional bool interQuery = 6 [default = false];
repeated Fetch fetches = 7;
optional bool shouldDie = 8;
+ optional KeyValueSetProto queryMeta = 9;
}
message Fetch {
@@ -104,7 +105,8 @@ message Partition {
message QueryExecutionRequestProto {
required QueryIdProto queryId = 1;
- required StringProto logicalPlanJson = 2;
+ required KeyValueSetProto queryMeta = 2;
+ required StringProto logicalPlanJson = 3;
}
message GetTaskRequestProto {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TajoTestingCluster.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TajoTestingCluster.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TajoTestingCluster.java
index f031938..ab8d4b4 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TajoTestingCluster.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/TajoTestingCluster.java
@@ -133,7 +133,7 @@ public class TajoTestingCluster {
String dirStr = getTestDir(randomStr).toString();
File dir = new File(dirStr).getAbsoluteFile();
// Have it cleaned up on exit
- //dir.deleteOnExit();
+ dir.deleteOnExit();
return dir;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/eval/TestEvalTree.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/eval/TestEvalTree.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/eval/TestEvalTree.java
index 2efaadc..e61a415 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/eval/TestEvalTree.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/eval/TestEvalTree.java
@@ -32,6 +32,7 @@ import org.apache.tajo.engine.json.CoreGsonHelper;
import org.apache.tajo.engine.parser.SQLAnalyzer;
import org.apache.tajo.engine.planner.LogicalPlan;
import org.apache.tajo.engine.planner.LogicalPlanner;
+import org.apache.tajo.engine.planner.PlanningException;
import org.apache.tajo.engine.planner.Target;
import org.apache.tajo.master.TajoMaster;
import org.apache.tajo.storage.Tuple;
@@ -124,7 +125,7 @@ public class TestEvalTree {
"select name from people where NOT (20 > 30)", // 5
};
- public static Target[] getRawTargets(String query) {
+ public static Target[] getRawTargets(String query) throws PlanningException {
Expr expr = analyzer.parse(query);
LogicalPlan plan = planner.createPlan(expr);
Target [] targets = plan.getRootBlock().getTargetListManager().getUnEvaluatedTargets();
@@ -136,7 +137,12 @@ public class TestEvalTree {
public static EvalNode getRootSelection(String query) {
Expr block = analyzer.parse(query);
- LogicalPlan plan = planner.createPlan(block);
+ LogicalPlan plan = null;
+ try {
+ plan = planner.createPlan(block);
+ } catch (PlanningException e) {
+ e.printStackTrace();
+ }
EvalNode qual = plan.getRootBlock().getSelectionNode().getQual();
assertJsonSerDer(qual);
return qual;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/eval/TestEvalTreeUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/eval/TestEvalTreeUtil.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/eval/TestEvalTreeUtil.java
index 1a24ec8..9b974cc 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/eval/TestEvalTreeUtil.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/eval/TestEvalTreeUtil.java
@@ -31,6 +31,7 @@ import org.apache.tajo.engine.eval.TestEvalTree.TestSum;
import org.apache.tajo.engine.parser.SQLAnalyzer;
import org.apache.tajo.engine.planner.LogicalPlan;
import org.apache.tajo.engine.planner.LogicalPlanner;
+import org.apache.tajo.engine.planner.PlanningException;
import org.apache.tajo.engine.planner.Target;
import org.apache.tajo.engine.planner.logical.EvalExprNode;
import org.apache.tajo.engine.planner.logical.NodeType;
@@ -96,7 +97,12 @@ public class TestEvalTreeUtil {
public static Target [] getRawTargets(String query) {
Expr expr = analyzer.parse(query);
- LogicalPlan plan = planner.createPlan(expr);
+ LogicalPlan plan = null;
+ try {
+ plan = planner.createPlan(expr);
+ } catch (PlanningException e) {
+ e.printStackTrace();
+ }
if (plan.getRootBlock().getRoot().getType() == NodeType.EXPRS) {
return ((EvalExprNode)plan.getRootBlock().getRoot()).getExprs();
} else {
@@ -106,7 +112,12 @@ public class TestEvalTreeUtil {
public static EvalNode getRootSelection(String query) {
Expr block = analyzer.parse(query);
- LogicalPlan plan = planner.createPlan(block);
+ LogicalPlan plan = null;
+ try {
+ plan = planner.createPlan(block);
+ } catch (PlanningException e) {
+ e.printStackTrace();
+ }
return plan.getRootBlock().getSelectionNode().getQual();
}
@@ -173,7 +184,7 @@ public class TestEvalTreeUtil {
}
@Test
- public final void testGetContainExprs() throws CloneNotSupportedException {
+ public final void testGetContainExprs() throws CloneNotSupportedException, PlanningException {
Expr expr = analyzer.parse(QUERIES[1]);
LogicalPlan plan = planner.createPlan(expr);
Target [] targets = plan.getRootBlock().getTargetListManager().getUnEvaluatedTargets();
@@ -236,7 +247,7 @@ public class TestEvalTreeUtil {
}
@Test
- public final void testSimplify() {
+ public final void testSimplify() throws PlanningException {
Target [] targets = getRawTargets(QUERIES[0]);
EvalNode node = AlgebraicUtil.simplify(targets[0].getEvalTree());
EvalContext nodeCtx = node.newContext();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/parser/TestSQLAnalyzer.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/parser/TestSQLAnalyzer.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/parser/TestSQLAnalyzer.java
index 5ba26f8..75a531f 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/parser/TestSQLAnalyzer.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/parser/TestSQLAnalyzer.java
@@ -78,18 +78,6 @@ public class TestSQLAnalyzer {
}
@Test
- public void testInsert1() throws IOException {
- String sql = FileUtil.readTextFile(new File("src/test/queries/insert_1.sql"));
- parseQuery(sql);
- }
-
- @Test
- public void testInsert2() throws IOException {
- String sql = FileUtil.readTextFile(new File("src/test/queries/insert_2.sql"));
- parseQuery(sql);
- }
-
- @Test
public void testGroupby1() throws IOException {
String sql = FileUtil.readTextFile(new File("src/test/queries/groupby_1.sql"));
parseQuery(sql);
@@ -251,6 +239,42 @@ public class TestSQLAnalyzer {
parseQuery(sql);
}
+ @Test
+ public void testInsertIntoTable() throws IOException {
+ String sql = FileUtil.readTextFile(new File("src/test/queries/insert_into_select_1.sql"));
+ parseQuery(sql);
+ }
+
+ @Test
+ public void testInsertIntoLocation() throws IOException {
+ String sql = FileUtil.readTextFile(new File("src/test/queries/insert_into_select_2.sql"));
+ parseQuery(sql);
+ }
+
+ @Test
+ public void testInsertIntoTable2() throws IOException {
+ String sql = FileUtil.readTextFile(new File("src/test/queries/insert_into_select_3.sql"));
+ parseQuery(sql);
+ }
+
+ @Test
+ public void testInsertOverwriteIntoTable() throws IOException {
+ String sql = FileUtil.readTextFile(new File("src/test/queries/insert_overwrite_into_select_1.sql"));
+ parseQuery(sql);
+ }
+
+ @Test
+ public void testInsertOverwriteIntoLocation() throws IOException {
+ String sql = FileUtil.readTextFile(new File("src/test/queries/insert_overwrite_into_select_2.sql"));
+ parseQuery(sql);
+ }
+
+ @Test
+ public void testInsertOverwriteIntoTable2() throws IOException {
+ String sql = FileUtil.readTextFile(new File("src/test/queries/insert_overwrite_into_select_3.sql"));
+ parseQuery(sql);
+ }
+
static String[] exprs = {
"1 + 2", // 0
"3 - 4", // 1
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestLogicalOptimizer.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestLogicalOptimizer.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestLogicalOptimizer.java
index 7c0b9d0..72cbfc2 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestLogicalOptimizer.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestLogicalOptimizer.java
@@ -242,4 +242,11 @@ public class TestLogicalOptimizer {
// Test for Join Node
assertTrue(PlannerUtil.canBeEvaluated(selNode.getQual(), scanNode));
}
+
+ @Test
+ public final void testInsertInto() throws CloneNotSupportedException, PlanningException {
+ Expr expr = sqlAnalyzer.parse(TestLogicalPlanner.insertStatements[0]);
+ LogicalPlan newPlan = planner.createPlan(expr);
+ optimizer.optimize(newPlan);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestLogicalPlanner.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestLogicalPlanner.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestLogicalPlanner.java
index 980f435..7f95a85 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestLogicalPlanner.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestLogicalPlanner.java
@@ -44,8 +44,7 @@ import java.io.File;
import java.io.IOException;
import java.util.*;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
public class TestLogicalPlanner {
private static TajoTestingCluster util;
@@ -62,7 +61,7 @@ public class TestLogicalPlanner {
for (FunctionDesc funcDesc : TajoMaster.initBuiltinFunctions()) {
catalog.registerFunction(funcDesc);
}
-
+
Schema schema = new Schema();
schema.addColumn("name", Type.TEXT);
schema.addColumn("empid", Type.INT4);
@@ -98,7 +97,7 @@ public class TestLogicalPlanner {
// TPC-H Schema for Complex Queries
String [] tpchTables = {
- "part", "supplier", "partsupp", "nation", "region"
+ "part", "supplier", "partsupp", "nation", "region", "lineitem"
};
tpch = new TPCH();
tpch.loadSchemas();
@@ -136,7 +135,7 @@ public class TestLogicalPlanner {
};
@Test
- public final void testSingleRelation() throws CloneNotSupportedException {
+ public final void testSingleRelation() throws CloneNotSupportedException, PlanningException {
Expr expr = sqlAnalyzer.parse(QUERIES[0]);
LogicalPlan planNode = planner.createPlan(expr);
LogicalNode plan = planNode.getRootBlock().getRoot();
@@ -168,7 +167,7 @@ public class TestLogicalPlanner {
}
@Test
- public final void testImplicityJoinPlan() throws CloneNotSupportedException {
+ public final void testImplicityJoinPlan() throws CloneNotSupportedException, PlanningException {
// two relations
Expr expr = sqlAnalyzer.parse(QUERIES[1]);
LogicalPlan planNode = planner.createPlan(expr);
@@ -264,7 +263,7 @@ public class TestLogicalPlanner {
}
@Test
- public final void testNaturalJoinPlan() {
+ public final void testNaturalJoinPlan() throws PlanningException {
// two relations
Expr context = sqlAnalyzer.parse(JOINS[0]);
LogicalNode plan = planner.createPlan(context).getRootBlock().getRoot();
@@ -300,7 +299,7 @@ public class TestLogicalPlanner {
}
@Test
- public final void testInnerJoinPlan() {
+ public final void testInnerJoinPlan() throws PlanningException {
// two relations
Expr expr = sqlAnalyzer.parse(JOINS[1]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
@@ -336,7 +335,7 @@ public class TestLogicalPlanner {
}
@Test
- public final void testOuterJoinPlan() {
+ public final void testOuterJoinPlan() throws PlanningException {
// two relations
Expr expr = sqlAnalyzer.parse(JOINS[2]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
@@ -374,7 +373,7 @@ public class TestLogicalPlanner {
@Test
- public final void testGroupby() throws CloneNotSupportedException {
+ public final void testGroupby() throws CloneNotSupportedException, PlanningException {
// without 'having clause'
Expr context = sqlAnalyzer.parse(QUERIES[7]);
LogicalNode plan = planner.createPlan(context).getRootBlock().getRoot();
@@ -412,7 +411,7 @@ public class TestLogicalPlanner {
@Test
- public final void testMultipleJoin() throws IOException {
+ public final void testMultipleJoin() throws IOException, PlanningException {
Expr expr = sqlAnalyzer.parse(
FileUtil.readTextFile(new File("src/test/queries/tpch_q2_simplified.tql")));
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
@@ -444,7 +443,7 @@ public class TestLogicalPlanner {
@Test
- public final void testStoreTable() throws CloneNotSupportedException {
+ public final void testStoreTable() throws CloneNotSupportedException, PlanningException {
Expr context = sqlAnalyzer.parse(QUERIES[8]);
LogicalNode plan = planner.createPlan(context).getRootBlock().getRoot();
TestLogicalNode.testCloneLogicalNode(plan);
@@ -460,7 +459,7 @@ public class TestLogicalPlanner {
}
@Test
- public final void testOrderBy() throws CloneNotSupportedException {
+ public final void testOrderBy() throws CloneNotSupportedException, PlanningException {
Expr expr = sqlAnalyzer.parse(QUERIES[4]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
testJsonSerDerObject(plan);
@@ -487,7 +486,7 @@ public class TestLogicalPlanner {
}
@Test
- public final void testLimit() throws CloneNotSupportedException {
+ public final void testLimit() throws CloneNotSupportedException, PlanningException {
Expr expr = sqlAnalyzer.parse(QUERIES[12]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
testJsonSerDerObject(plan);
@@ -506,7 +505,7 @@ public class TestLogicalPlanner {
}
@Test
- public final void testSPJPush() throws CloneNotSupportedException {
+ public final void testSPJPush() throws CloneNotSupportedException, PlanningException {
Expr expr = sqlAnalyzer.parse(QUERIES[5]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
testJsonSerDerObject(plan);
@@ -535,7 +534,7 @@ public class TestLogicalPlanner {
@Test
- public final void testSPJ() throws CloneNotSupportedException {
+ public final void testSPJ() throws CloneNotSupportedException, PlanningException {
Expr expr = sqlAnalyzer.parse(QUERIES[6]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
testJsonSerDerObject(plan);
@@ -543,7 +542,7 @@ public class TestLogicalPlanner {
}
@Test
- public final void testJson() {
+ public final void testJson() throws PlanningException {
Expr expr = sqlAnalyzer.parse(QUERIES[9]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
testJsonSerDerObject(plan);
@@ -560,7 +559,7 @@ public class TestLogicalPlanner {
}
@Test
- public final void testVisitor() {
+ public final void testVisitor() throws PlanningException {
// two relations
Expr expr = sqlAnalyzer.parse(QUERIES[1]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
@@ -585,7 +584,7 @@ public class TestLogicalPlanner {
@Test
- public final void testExprNode() {
+ public final void testExprNode() throws PlanningException {
Expr expr = sqlAnalyzer.parse(QUERIES[10]);
LogicalPlan rootNode = planner.createPlan(expr);
LogicalNode plan = rootNode.getRootBlock().getRoot();
@@ -611,7 +610,7 @@ public class TestLogicalPlanner {
@Test
- public final void testAlias1() {
+ public final void testAlias1() throws PlanningException {
Expr expr = sqlAnalyzer.parse(ALIAS[0]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
LogicalRootNode root = (LogicalRootNode) plan;
@@ -638,7 +637,7 @@ public class TestLogicalPlanner {
}
@Test
- public final void testAlias2() {
+ public final void testAlias2() throws PlanningException {
Expr expr = sqlAnalyzer.parse(ALIAS[1]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
LogicalRootNode root = (LogicalRootNode) plan;
@@ -657,7 +656,7 @@ public class TestLogicalPlanner {
};
@Test
- public final void testCreateTableDef() {
+ public final void testCreateTableDef() throws PlanningException {
Expr expr = sqlAnalyzer.parse(CREATE_TABLE[0]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
LogicalRootNode root = (LogicalRootNode) plan;
@@ -741,7 +740,7 @@ public class TestLogicalPlanner {
};
@Test
- public final void testCubeBy() {
+ public final void testCubeBy() throws PlanningException {
Expr expr = sqlAnalyzer.parse(CUBE_ROLLUP[0]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
testJsonSerDerObject(plan);
@@ -778,7 +777,6 @@ public class TestLogicalPlanner {
}
}
-
static final String setStatements [] = {
"select deptName from employee where deptName like 'data%' union select deptName from score where deptName like 'data%'",
"select deptName from employee union select deptName from score as s1 intersect select deptName from score as s2",
@@ -786,7 +784,7 @@ public class TestLogicalPlanner {
};
@Test
- public final void testSetPlan() {
+ public final void testSetPlan() throws PlanningException {
Expr expr = sqlAnalyzer.parse(setStatements[0]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
testJsonSerDerObject(plan);
@@ -803,7 +801,7 @@ public class TestLogicalPlanner {
}
@Test
- public final void testSetPlan2() {
+ public final void testSetPlan2() throws PlanningException {
// for testing multiple set statements
Expr expr = sqlAnalyzer.parse(setStatements[1]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
@@ -820,7 +818,7 @@ public class TestLogicalPlanner {
}
@Test
- public final void testSetPlan3() {
+ public final void testSetPlan3() throws PlanningException {
// for testing multiple set statements
Expr expr = sqlAnalyzer.parse(setStatements[2]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
@@ -839,8 +837,6 @@ public class TestLogicalPlanner {
assertEquals(NodeType.PROJECTION, intersect.getRightChild().getType());
}
-
-
static final String [] setQualifiers = {
"select name, empid from employee",
"select distinct name, empid from employee",
@@ -848,7 +844,7 @@ public class TestLogicalPlanner {
};
@Test
- public void testSetQualifier() {
+ public void testSetQualifier() throws PlanningException {
Expr context = sqlAnalyzer.parse(setQualifiers[0]);
LogicalNode plan = planner.createPlan(context).getRootBlock().getRoot();
testJsonSerDerObject(plan);
@@ -881,4 +877,93 @@ public class TestLogicalPlanner {
LogicalNode fromJson = CoreGsonHelper.fromJson(json, LogicalNode.class);
assertEquals("JSON (de) serialization equivalence check", rootNode, fromJson);
}
+
+ // Table descriptions
+ //
+ // employee (name text, empid int4, deptname text)
+ // dept (deptname text, nameger text)
+ // score (deptname text, score inet4)
+
+ static final String [] insertStatements = {
+ "insert into score select name from employee", // 0
+ "insert into score select name, empid from employee", // 1
+ "insert into employee (name, deptname) select * from dept", // 2
+ "insert into location '/tmp/data' select name, empid from employee", // 3
+ "insert overwrite into employee (name, deptname) select * from dept", // 4
+ "insert overwrite into LOCATION '/tmp/data' select * from dept" // 5
+ };
+
+ @Test
+ public final void testInsertInto0() throws PlanningException {
+ Expr expr = sqlAnalyzer.parse(insertStatements[0]);
+ LogicalPlan plan = planner.createPlan(expr);
+ assertEquals(2, plan.getQueryBlocks().size());
+ InsertNode insertNode = getInsertNode(plan);
+ assertFalse(insertNode.isOverwrite());
+ assertTrue(insertNode.hasTargetTable());
+ assertEquals("score", insertNode.getTargetTable().getName());
+ }
+
+ @Test
+ public final void testInsertInto1() throws PlanningException {
+ Expr expr = sqlAnalyzer.parse(insertStatements[1]);
+ LogicalPlan plan = planner.createPlan(expr);
+ assertEquals(2, plan.getQueryBlocks().size());
+ InsertNode insertNode = getInsertNode(plan);
+ assertFalse(insertNode.isOverwrite());
+ assertEquals("score", insertNode.getTargetTable().getName());
+ }
+
+ @Test
+ public final void testInsertInto2() throws PlanningException {
+ Expr expr = sqlAnalyzer.parse(insertStatements[2]);
+ LogicalPlan plan = planner.createPlan(expr);
+ assertEquals(2, plan.getQueryBlocks().size());
+ InsertNode insertNode = getInsertNode(plan);
+ assertFalse(insertNode.isOverwrite());
+ assertEquals("employee", insertNode.getTargetTable().getName());
+ assertTrue(insertNode.hasTargetSchema());
+ assertEquals(insertNode.getTargetSchema().getColumn(0).getColumnName(), "name");
+ assertEquals(insertNode.getTargetSchema().getColumn(1).getColumnName(), "deptname");
+ }
+
+ @Test
+ public final void testInsertInto3() throws PlanningException {
+ Expr expr = sqlAnalyzer.parse(insertStatements[3]);
+ LogicalPlan plan = planner.createPlan(expr);
+ assertEquals(2, plan.getQueryBlocks().size());
+ InsertNode insertNode = getInsertNode(plan);
+ assertFalse(insertNode.isOverwrite());
+ assertTrue(insertNode.hasPath());
+ }
+
+ @Test
+ public final void testInsertInto4() throws PlanningException {
+ Expr expr = sqlAnalyzer.parse(insertStatements[4]);
+ LogicalPlan plan = planner.createPlan(expr);
+ assertEquals(2, plan.getQueryBlocks().size());
+ InsertNode insertNode = getInsertNode(plan);
+ assertTrue(insertNode.isOverwrite());
+ assertTrue(insertNode.hasTargetTable());
+ assertEquals("employee", insertNode.getTargetTable().getName());
+ assertTrue(insertNode.hasTargetSchema());
+ assertEquals(insertNode.getTargetSchema().getColumn(0).getColumnName(), "name");
+ assertEquals(insertNode.getTargetSchema().getColumn(1).getColumnName(), "deptname");
+ }
+
+ @Test
+ public final void testInsertInto5() throws PlanningException {
+ Expr expr = sqlAnalyzer.parse(insertStatements[5]);
+ LogicalPlan plan = planner.createPlan(expr);
+ assertEquals(2, plan.getQueryBlocks().size());
+ InsertNode insertNode = getInsertNode(plan);
+ assertTrue(insertNode.isOverwrite());
+ assertTrue(insertNode.hasPath());
+ }
+
+ private static InsertNode getInsertNode(LogicalPlan plan) {
+ LogicalRootNode root = (LogicalRootNode) plan.getRootBlock().getRoot();
+ assertEquals(NodeType.INSERT, root.getChild().getType());
+ return (InsertNode) root.getChild();
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestPlannerUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestPlannerUtil.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestPlannerUtil.java
index 1e95355..4821751 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestPlannerUtil.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/TestPlannerUtil.java
@@ -96,7 +96,7 @@ public class TestPlannerUtil {
}
@Test
- public final void testTransformTwoPhase() {
+ public final void testTransformTwoPhase() throws PlanningException {
// without 'having clause'
Expr expr = analyzer.parse(TestLogicalPlanner.QUERIES[7]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
@@ -111,7 +111,7 @@ public class TestPlannerUtil {
}
@Test
- public final void testTrasformTwoPhaseWithStore() {
+ public final void testTrasformTwoPhaseWithStore() throws PlanningException {
Expr expr = analyzer.parse(TestLogicalPlanner.QUERIES[9]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
@@ -141,7 +141,7 @@ public class TestPlannerUtil {
}
@Test
- public final void testFindTopNode() throws CloneNotSupportedException {
+ public final void testFindTopNode() throws CloneNotSupportedException, PlanningException {
// two relations
Expr expr = analyzer.parse(TestLogicalPlanner.QUERIES[1]);
LogicalNode plan = planner.createPlan(expr).getRootBlock().getRoot();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestBNLJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestBNLJoinExec.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestBNLJoinExec.java
index bea1eb6..97459f9 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestBNLJoinExec.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestBNLJoinExec.java
@@ -32,6 +32,7 @@ import org.apache.tajo.engine.parser.SQLAnalyzer;
import org.apache.tajo.engine.planner.LogicalPlanner;
import org.apache.tajo.engine.planner.PhysicalPlanner;
import org.apache.tajo.engine.planner.PhysicalPlannerImpl;
+import org.apache.tajo.engine.planner.PlanningException;
import org.apache.tajo.engine.planner.logical.JoinNode;
import org.apache.tajo.engine.planner.logical.LogicalNode;
import org.apache.tajo.storage.*;
@@ -129,7 +130,7 @@ public class TestBNLJoinExec {
"inner join people as p on e.empId = p.empId and e.memId = p.fk_memId" };
@Test
- public final void testBNLCrossJoin() throws IOException {
+ public final void testBNLCrossJoin() throws IOException, PlanningException {
Fragment[] empFrags = StorageManager.splitNG(conf, "employee", employee.getMeta(), employee.getPath(),
Integer.MAX_VALUE);
Fragment[] peopleFrags = StorageManager.splitNG(conf, "people", people.getMeta(), people.getPath(),
@@ -164,7 +165,7 @@ public class TestBNLJoinExec {
}
@Test
- public final void testBNLInnerJoin() throws IOException {
+ public final void testBNLInnerJoin() throws IOException, PlanningException {
Fragment[] empFrags = StorageManager.splitNG(conf, "employee", employee.getMeta(), employee.getPath(),
Integer.MAX_VALUE);
Fragment[] peopleFrags = StorageManager.splitNG(conf, "people", people.getMeta(), people.getPath(),
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java
index 621f5a2..0fc3773 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestExternalSortExec.java
@@ -29,10 +29,7 @@ import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.datum.Datum;
import org.apache.tajo.datum.DatumFactory;
import org.apache.tajo.engine.parser.SQLAnalyzer;
-import org.apache.tajo.engine.planner.LogicalPlan;
-import org.apache.tajo.engine.planner.LogicalPlanner;
-import org.apache.tajo.engine.planner.PhysicalPlanner;
-import org.apache.tajo.engine.planner.PhysicalPlannerImpl;
+import org.apache.tajo.engine.planner.*;
import org.apache.tajo.engine.planner.logical.LogicalNode;
import org.apache.tajo.storage.*;
import org.apache.tajo.util.CommonTestingUtil;
@@ -110,7 +107,7 @@ public class TestExternalSortExec {
};
@Test
- public final void testNext() throws IOException {
+ public final void testNext() throws IOException, PlanningException {
Fragment[] frags = sm.splitNG(conf, "employee", employee.getMeta(), employee.getPath(),
Integer.MAX_VALUE);
Path workDir = new Path(testDir, TestExternalSortExec.class.getName());
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestHashJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestHashJoinExec.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestHashJoinExec.java
index 9422358..e270df3 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestHashJoinExec.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestHashJoinExec.java
@@ -32,6 +32,7 @@ import org.apache.tajo.engine.parser.SQLAnalyzer;
import org.apache.tajo.engine.planner.LogicalPlanner;
import org.apache.tajo.engine.planner.PhysicalPlanner;
import org.apache.tajo.engine.planner.PhysicalPlannerImpl;
+import org.apache.tajo.engine.planner.PlanningException;
import org.apache.tajo.engine.planner.logical.LogicalNode;
import org.apache.tajo.storage.*;
import org.apache.tajo.util.CommonTestingUtil;
@@ -129,7 +130,7 @@ public class TestHashJoinExec {
};
@Test
- public final void testHashInnerJoin() throws IOException {
+ public final void testHashInnerJoin() throws IOException, PlanningException {
Fragment[] empFrags = StorageManager.splitNG(conf, "employee", employee.getMeta(), employee.getPath(),
Integer.MAX_VALUE);
Fragment[] peopleFrags = StorageManager.splitNG(conf, "people", people.getMeta(), people.getPath(),
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestMergeJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestMergeJoinExec.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestMergeJoinExec.java
index 9d85970..776882b 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestMergeJoinExec.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestMergeJoinExec.java
@@ -33,6 +33,7 @@ import org.apache.tajo.engine.parser.SQLAnalyzer;
import org.apache.tajo.engine.planner.LogicalPlanner;
import org.apache.tajo.engine.planner.PhysicalPlanner;
import org.apache.tajo.engine.planner.PhysicalPlannerImpl;
+import org.apache.tajo.engine.planner.PlanningException;
import org.apache.tajo.engine.planner.logical.LogicalNode;
import org.apache.tajo.engine.planner.logical.SortNode;
import org.apache.tajo.storage.*;
@@ -145,7 +146,7 @@ public class TestMergeJoinExec {
};
@Test
- public final void testMergeInnerJoin() throws IOException {
+ public final void testMergeInnerJoin() throws IOException, PlanningException {
Fragment[] empFrags = sm.splitNG(conf, "employee", employee.getMeta(), employee.getPath(),
Integer.MAX_VALUE);
Fragment[] peopleFrags = sm.splitNG(conf, "people", people.getMeta(), people.getPath(),
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestNLJoinExec.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestNLJoinExec.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestNLJoinExec.java
index 7235924..9289dc9 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestNLJoinExec.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/planner/physical/TestNLJoinExec.java
@@ -19,24 +19,27 @@
package org.apache.tajo.engine.planner.physical;
import org.apache.hadoop.fs.Path;
-import org.apache.tajo.algebra.Expr;
-import org.apache.tajo.engine.parser.SQLAnalyzer;
-import org.apache.tajo.engine.planner.*;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
import org.apache.tajo.TajoTestingCluster;
import org.apache.tajo.TaskAttemptContext;
+import org.apache.tajo.algebra.Expr;
import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
import org.apache.tajo.common.TajoDataTypes.Type;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.datum.Datum;
import org.apache.tajo.datum.DatumFactory;
+import org.apache.tajo.engine.parser.SQLAnalyzer;
+import org.apache.tajo.engine.planner.LogicalPlanner;
+import org.apache.tajo.engine.planner.PhysicalPlanner;
+import org.apache.tajo.engine.planner.PhysicalPlannerImpl;
+import org.apache.tajo.engine.planner.PlanningException;
import org.apache.tajo.engine.planner.logical.LogicalNode;
import org.apache.tajo.storage.*;
import org.apache.tajo.util.CommonTestingUtil;
import org.apache.tajo.util.TUtil;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
import java.io.IOException;
@@ -129,7 +132,7 @@ public class TestNLJoinExec {
};
@Test
- public final void testNLCrossJoin() throws IOException {
+ public final void testNLCrossJoin() throws IOException, PlanningException {
Fragment[] empFrags = StorageManager.splitNG(conf, "employee", employee.getMeta(), employee.getPath(),
Integer.MAX_VALUE);
Fragment[] peopleFrags = StorageManager.splitNG(conf, "people", people.getMeta(), people.getPath(),
@@ -156,7 +159,7 @@ public class TestNLJoinExec {
}
@Test
- public final void testNLInnerJoin() throws IOException {
+ public final void testNLInnerJoin() throws IOException, PlanningException {
Fragment[] empFrags = StorageManager.splitNG(conf, "employee", employee.getMeta(), employee.getPath(),
Integer.MAX_VALUE);
Fragment[] peopleFrags = StorageManager.splitNG(conf, "people", people.getMeta(), people.getPath(),
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestInsertQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestInsertQuery.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestInsertQuery.java
new file mode 100644
index 0000000..e9a0c65
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestInsertQuery.java
@@ -0,0 +1,180 @@
+/**
+ * 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.tajo.engine.query;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.IntegrationTest;
+import org.apache.tajo.TajoTestingCluster;
+import org.apache.tajo.TpchTestBase;
+import org.apache.tajo.catalog.CatalogService;
+import org.apache.tajo.catalog.TableDesc;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.io.IOException;
+import java.sql.ResultSet;
+
+import static org.junit.Assert.*;
+
+@Category(IntegrationTest.class)
+public class TestInsertQuery {
+ private static TpchTestBase tpch;
+ public TestInsertQuery() throws IOException {
+ super();
+ }
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ tpch = TpchTestBase.getInstance();
+ }
+
+ @Test
+ public final void testInsertOverwrite() throws Exception {
+ String tableName ="InsertOverwrite";
+ tpch.execute("create table " + tableName +" (col1 int8, col2 int4, col3 float4)");
+ TajoTestingCluster cluster = tpch.getTestingCluster();
+ CatalogService catalog = cluster.getMaster().getCatalog();
+ assertTrue(catalog.existsTable(tableName));
+
+ tpch.execute("insert overwrite into " + tableName + " select l_orderkey, l_partkey, l_quantity from lineitem");
+ TableDesc desc = catalog.getTableDesc(tableName);
+ assertEquals(5, desc.getMeta().getStat().getNumRows().intValue());
+ }
+
+ @Test
+ public final void testInsertOverwriteSmallerColumns() throws Exception {
+ String tableName = "insertoverwritesmallercolumns";
+ tpch.execute("create table " + tableName + " (col1 int8, col2 int4, col3 float4)");
+ TajoTestingCluster cluster = tpch.getTestingCluster();
+ CatalogService catalog = cluster.getMaster().getCatalog();
+ assertTrue(catalog.existsTable(tableName));
+ TableDesc originalDesc = catalog.getTableDesc(tableName);
+
+ tpch.execute("insert overwrite into " + tableName + " select l_orderkey from lineitem");
+ TableDesc desc = catalog.getTableDesc(tableName);
+ assertEquals(5, desc.getMeta().getStat().getNumRows().intValue());
+ assertEquals(originalDesc.getMeta().getSchema(), desc.getMeta().getSchema());
+ }
+
+ @Test
+ public final void testInsertOverwriteWithTargetColumns() throws Exception {
+ String tableName = "InsertOverwriteWithTargetColumns";
+ tpch.execute("create table " + tableName + " (col1 int8, col2 int4, col3 float4)");
+ TajoTestingCluster cluster = tpch.getTestingCluster();
+ CatalogService catalog = cluster.getMaster().getCatalog();
+ assertTrue(catalog.existsTable(tableName));
+ TableDesc originalDesc = catalog.getTableDesc(tableName);
+
+ tpch.execute("insert overwrite into " + tableName + " (col1, col3) select l_orderkey, l_quantity from lineitem");
+ TableDesc desc = catalog.getTableDesc(tableName);
+ assertEquals(5, desc.getMeta().getStat().getNumRows().intValue());
+
+ ResultSet res = tpch.execute("select * from " + tableName);
+ assertTrue(res.next());
+ assertEquals(1, res.getLong(1));
+ assertEquals("null", res.getString(2));
+ assertTrue(17.0 == res.getFloat(3));
+
+ assertTrue(res.next());
+ assertEquals(1, res.getLong(1));
+ assertEquals("null", res.getString(2));
+ assertTrue(36.0 == res.getFloat(3));
+
+ assertTrue(res.next());
+ assertEquals(2, res.getLong(1));
+ assertEquals("null", res.getString(2));
+ assertTrue(38.0 == res.getFloat(3));
+
+ assertTrue(res.next());
+ assertEquals(3, res.getLong(1));
+ assertEquals("null", res.getString(2));
+ assertTrue(45.0 == res.getFloat(3));
+
+ assertTrue(res.next());
+ assertEquals(3, res.getLong(1));
+ assertEquals("null", res.getString(2));
+ assertTrue(49.0 == res.getFloat(3));
+
+ assertFalse(res.next());
+ res.close();
+
+ assertEquals(originalDesc.getMeta().getSchema(), desc.getMeta().getSchema());
+ }
+
+ @Test
+ public final void testInsertOverwriteWithAsterisk() throws Exception {
+ String tableName = "testinsertoverwritewithasterisk";
+ tpch.execute("create table " + tableName + " as select * from lineitem");
+ TajoTestingCluster cluster = tpch.getTestingCluster();
+ CatalogService catalog = cluster.getMaster().getCatalog();
+ assertTrue(catalog.existsTable(tableName));
+
+ tpch.execute("insert overwrite into " + tableName + " select * from lineitem where l_orderkey = 3");
+ TableDesc desc = catalog.getTableDesc(tableName);
+ assertEquals(2, desc.getMeta().getStat().getNumRows().intValue());
+ }
+
+ @Test
+ public final void testInsertOverwriteIntoSelect() throws Exception {
+ String tableName = "insertoverwriteintoselect";
+ ResultSet res = tpch.execute(
+ "create table " + tableName + " as select l_orderkey from lineitem");
+ assertFalse(res.next());
+ res.close();
+
+ TajoTestingCluster cluster = tpch.getTestingCluster();
+ CatalogService catalog = cluster.getMaster().getCatalog();
+ assertTrue(catalog.existsTable(tableName));
+ TableDesc orderKeys = catalog.getTableDesc(tableName);
+ assertEquals(5, orderKeys.getMeta().getStat().getNumRows().intValue());
+
+ // this query will result in the two rows.
+ res = tpch.execute(
+ "insert overwrite into " + tableName + " select l_orderkey from lineitem where l_orderkey = 3");
+ assertFalse(res.next());
+ res.close();
+
+ assertTrue(catalog.existsTable(tableName));
+ orderKeys = catalog.getTableDesc(tableName);
+ assertEquals(2, orderKeys.getMeta().getStat().getNumRows().intValue());
+ }
+
+ @Test
+ public final void testInsertOverwriteCapitalTableName() throws Exception {
+ String tableName = "testInsertOverwriteCapitalTableName";
+ tpch.execute("create table " + tableName + " as select * from lineitem");
+ TajoTestingCluster cluster = tpch.getTestingCluster();
+ CatalogService catalog = cluster.getMaster().getCatalog();
+ assertTrue(catalog.existsTable(tableName));
+
+ tpch.execute("insert overwrite into " + tableName + " select * from lineitem where l_orderkey = 3");
+ TableDesc desc = catalog.getTableDesc(tableName);
+ assertEquals(2, desc.getMeta().getStat().getNumRows().intValue());
+ }
+
+ @Test
+ public final void testInsertOverwriteLocation() throws Exception {
+ tpch.execute("insert overwrite into location '/tajo-data/testInsertOverwriteCapitalTableName' select * from lineitem where l_orderkey = 3");
+ FileSystem fs = FileSystem.get(tpch.getTestingCluster().getConfiguration());
+ assertTrue(fs.exists(new Path("/tajo-data/testInsertOverwriteCapitalTableName")));
+ assertEquals(1, fs.listStatus(new Path("/tajo-data/testInsertOverwriteCapitalTableName")).length);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestSelectQuery.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestSelectQuery.java b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestSelectQuery.java
index e96f7e5..f2a2a71 100644
--- a/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestSelectQuery.java
+++ b/tajo-core/tajo-core-backend/src/test/java/org/apache/tajo/engine/query/TestSelectQuery.java
@@ -21,7 +21,10 @@ package org.apache.tajo.engine.query;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.apache.tajo.IntegrationTest;
+import org.apache.tajo.TajoTestingCluster;
import org.apache.tajo.TpchTestBase;
+import org.apache.tajo.catalog.CatalogService;
+import org.apache.tajo.catalog.TableDesc;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -44,7 +47,7 @@ public class TestSelectQuery {
public static void setUp() throws Exception {
tpch = TpchTestBase.getInstance();
}
-
+
@Test
public final void testSelect() throws Exception {
ResultSet res = tpch.execute("select l_orderkey, l_partkey from lineitem");
@@ -277,7 +280,7 @@ public class TestSelectQuery {
try {
Map<Integer, String> result = Maps.newHashMap();
- result.put(0, "NULL");
+ result.put(0, "null");
result.put(1, "one");
result.put(2, "two");
result.put(3, "three");
@@ -406,15 +409,12 @@ public class TestSelectQuery {
public final void testCreateAfterSelect() throws Exception {
ResultSet res = tpch.execute(
"create table orderkeys as select l_orderkey from lineitem");
- try {
- int count = 0;
- for (;res.next();) {
- count++;
- }
- assertEquals(count, 5);
- } finally {
- res.close();
- }
+ res.close();
+ TajoTestingCluster cluster = tpch.getTestingCluster();
+ CatalogService catalog = cluster.getMaster().getCatalog();
+ assertTrue(catalog.existsTable("orderkeys"));
+ TableDesc orderKeys = catalog.getTableDesc("orderkeys");
+ assertEquals(5, orderKeys.getMeta().getStat().getNumRows().intValue());
}
//@Test
[3/3] git commit: TAJO-144: Implement INSERT OVERWRITE clause.
(hyunsik)
Posted by hy...@apache.org.
TAJO-144: Implement INSERT OVERWRITE clause. (hyunsik)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tajo/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tajo/commit/ffdb1c71
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tajo/tree/ffdb1c71
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tajo/diff/ffdb1c71
Branch: refs/heads/master
Commit: ffdb1c71cf796adafb635b99e80b8764a7b201ea
Parents: b2dc84f
Author: Hyunsik Choi <hy...@apache.org>
Authored: Thu Sep 5 00:50:35 2013 +0900
Committer: Hyunsik Choi <hy...@apache.org>
Committed: Thu Sep 5 00:50:35 2013 +0900
----------------------------------------------------------------------
.gitignore | 1 +
CHANGES.txt | 2 +
.../java/org/apache/tajo/algebra/Insert.java | 124 ++++++++
.../java/org/apache/tajo/algebra/OpType.java | 3 +
.../org/apache/tajo/catalog/CatalogClient.java | 24 +-
.../apache/tajo/catalog/MiniCatalogServer.java | 4 +
.../org/apache/tajo/catalog/store/DBStore.java | 7 +-
.../org/apache/tajo/datum/DatumFactory.java | 2 +
.../java/org/apache/tajo/datum/NullDatum.java | 6 +-
tajo-common/src/main/proto/DataTypes.proto | 1 +
.../org/apache/tajo/engine/parser/SQLLexer.g4 | 1 +
.../org/apache/tajo/engine/parser/SQLParser.g4 | 5 +-
.../main/java/org/apache/tajo/QueryConf.java | 27 --
.../main/java/org/apache/tajo/cli/TajoCli.java | 80 +++---
.../org/apache/tajo/engine/eval/ConstEval.java | 1 +
.../apache/tajo/engine/parser/SQLAnalyzer.java | 43 +++
.../tajo/engine/planner/BaseAlgebraVisitor.java | 10 +
.../engine/planner/BasicLogicalPlanVisitor.java | 8 +
.../apache/tajo/engine/planner/InsertNode.java | 183 ++++++++++++
.../apache/tajo/engine/planner/LogicalPlan.java | 79 +++++-
.../tajo/engine/planner/LogicalPlanVisitor.java | 2 +
.../tajo/engine/planner/LogicalPlanner.java | 77 ++++-
.../apache/tajo/engine/planner/PlannerUtil.java | 4 +
.../tajo/engine/planner/QueryBlockGraph.java | 91 ++++++
.../tajo/engine/planner/logical/NodeType.java | 3 +
.../engine/planner/logical/StoreTableNode.java | 24 +-
.../planner/rewrite/FilterPushDownRule.java | 26 +-
.../planner/rewrite/ProjectionPushDownRule.java | 62 ++--
.../tajo/engine/query/QueryUnitRequestImpl.java | 31 +-
.../ipc/protocolrecords/QueryUnitRequest.java | 2 +
.../org/apache/tajo/master/GlobalEngine.java | 281 ++++++++++++++-----
.../java/org/apache/tajo/master/QueryMeta.java | 118 ++++++++
.../apache/tajo/master/TaskSchedulerImpl.java | 7 +-
.../tajo/master/event/QueryStartEvent.java | 12 +-
.../apache/tajo/master/querymaster/Query.java | 36 ++-
.../master/querymaster/QueryInProgress.java | 8 +-
.../master/querymaster/QueryJobManager.java | 5 +-
.../tajo/master/querymaster/QueryMaster.java | 9 +-
.../master/querymaster/QueryMasterTask.java | 47 ++--
.../tajo/master/querymaster/SubQuery.java | 6 +-
.../tajo/worker/TajoResourceAllocator.java | 2 +-
.../java/org/apache/tajo/worker/TajoWorker.java | 53 +++-
.../tajo/worker/TajoWorkerClientService.java | 9 +-
.../tajo/worker/TajoWorkerManagerService.java | 6 +-
.../main/java/org/apache/tajo/worker/Task.java | 5 +-
.../java/org/apache/tajo/worker/TaskRunner.java | 18 --
.../src/main/proto/TajoWorkerProtocol.proto | 4 +-
.../org/apache/tajo/TajoTestingCluster.java | 2 +-
.../apache/tajo/engine/eval/TestEvalTree.java | 10 +-
.../tajo/engine/eval/TestEvalTreeUtil.java | 19 +-
.../tajo/engine/parser/TestSQLAnalyzer.java | 48 +++-
.../engine/planner/TestLogicalOptimizer.java | 7 +
.../tajo/engine/planner/TestLogicalPlanner.java | 145 ++++++++--
.../tajo/engine/planner/TestPlannerUtil.java | 6 +-
.../planner/physical/TestBNLJoinExec.java | 5 +-
.../planner/physical/TestExternalSortExec.java | 7 +-
.../planner/physical/TestHashJoinExec.java | 3 +-
.../planner/physical/TestMergeJoinExec.java | 3 +-
.../engine/planner/physical/TestNLJoinExec.java | 19 +-
.../tajo/engine/query/TestInsertQuery.java | 180 ++++++++++++
.../tajo/engine/query/TestSelectQuery.java | 22 +-
.../org/apache/tajo/worker/TaskRunnerTest.java | 64 -----
.../src/test/queries/insert_1.sql | 1 -
.../src/test/queries/insert_2.sql | 1 -
.../src/test/queries/insert_into_select_1.sql | 1 +
.../src/test/queries/insert_into_select_2.sql | 1 +
.../src/test/queries/insert_into_select_3.sql | 1 +
.../queries/insert_overwrite_into_select_1.sql | 1 +
.../queries/insert_overwrite_into_select_2.sql | 1 +
.../queries/insert_overwrite_into_select_3.sql | 1 +
.../java/org/apache/tajo/storage/CSVFile.java | 5 +-
.../java/org/apache/tajo/storage/RawFile.java | 19 +-
.../java/org/apache/tajo/storage/RowFile.java | 3 +
.../org/apache/tajo/storage/StorageManager.java | 19 ++
.../tajo/storage/rcfile/RCFileWrapper.java | 5 +
.../tajo/storage/trevni/TrevniAppender.java | 12 +-
.../tajo/storage/trevni/TrevniScanner.java | 16 +-
.../org/apache/tajo/storage/TestStorages.java | 12 +-
78 files changed, 1737 insertions(+), 461 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/.gitignore
----------------------------------------------------------------------
diff --git a/.gitignore b/.gitignore
index 4c76186..a780a50 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,3 +7,4 @@ target
*.iml
derby.log
.reviewboardrc
+atlassian-ide-plugin.xml
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index c16c833..1e8edbb 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -4,6 +4,8 @@ Release 0.2.0 - unreleased
NEW FEATURES
+ TAJO-144: Implement INSERT OVERWRITE clause. (hyunsik)
+
TAJO-139: TAJO-139: In predicate support. (hyunsik)
TAJO-134: Support for compression/decompression of CSVFile. (jinho)
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-algebra/src/main/java/org/apache/tajo/algebra/Insert.java
----------------------------------------------------------------------
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/Insert.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Insert.java
new file mode 100644
index 0000000..38c7ad4
--- /dev/null
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/Insert.java
@@ -0,0 +1,124 @@
+/**
+ * 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.tajo.algebra;
+
+import org.apache.tajo.util.TUtil;
+
+import java.util.Map;
+
+public class Insert extends Expr {
+ private boolean overwrite = false;
+ private String tableName;
+ private String [] targetColumns;
+ private String storageType;
+ private String location;
+ private Expr subquery;
+ private Map<String, String> params;
+
+ public Insert() {
+ super(OpType.Insert);
+ }
+
+ public void setOverwrite() {
+ overwrite = true;
+ }
+
+ public boolean isOverwrite() {
+ return overwrite;
+ }
+
+ public boolean hasTableName() {
+ return this.tableName != null;
+ }
+
+ public void setTableName(String tableName) {
+ this.tableName = tableName;
+ }
+
+ public String getTableName() {
+ return this.tableName;
+ }
+
+ public boolean hasTargetColumns() {
+ return targetColumns != null;
+ }
+
+ public String [] getTargetColumns() {
+ return targetColumns;
+ }
+
+ public void setTargetColumns(String [] targets) {
+ this.targetColumns = targets;
+ }
+
+ public boolean hasLocation() {
+ return location != null;
+ }
+
+ public String getLocation() {
+ return this.location;
+ }
+
+ public void setLocation(String location) {
+ this.location = location;
+ }
+
+ public boolean hasStorageType() {
+ return storageType != null;
+ }
+
+ public void setStorageType(String storageType) {
+ this.storageType = storageType;
+ }
+
+ public String getStorageType() {
+ return storageType;
+ }
+
+ public boolean hasParams() {
+ return params != null;
+ }
+
+ public Map<String, String> getParams() {
+ return params;
+ }
+
+ public void setParams(Map<String, String> params) {
+ this.params = params;
+ }
+
+ public void setSubQuery(Expr subquery) {
+ this.subquery = subquery;
+ }
+
+ public Expr getSubQuery() {
+ return subquery;
+ }
+
+ @Override
+ boolean equalsTo(Expr expr) {
+ Insert another = (Insert) expr;
+ return overwrite == another.overwrite &&
+ TUtil.checkEquals(tableName, another.tableName) &&
+ TUtil.checkEquals(subquery, another.subquery) &&
+ TUtil.checkEquals(storageType, another.storageType) &&
+ TUtil.checkEquals(location, another.location) &&
+ TUtil.checkEquals(params, another.params);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-algebra/src/main/java/org/apache/tajo/algebra/OpType.java
----------------------------------------------------------------------
diff --git a/tajo-algebra/src/main/java/org/apache/tajo/algebra/OpType.java b/tajo-algebra/src/main/java/org/apache/tajo/algebra/OpType.java
index 7be2ea9..6504a91 100644
--- a/tajo-algebra/src/main/java/org/apache/tajo/algebra/OpType.java
+++ b/tajo-algebra/src/main/java/org/apache/tajo/algebra/OpType.java
@@ -44,6 +44,9 @@ public enum OpType {
CreateTable(CreateTable.class),
DropTable(DropTable.class),
+ // Insert or Update
+ Insert(Insert.class),
+
// logical operators
And(BinaryOperator.class),
Or(BinaryOperator.class),
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-catalog/tajo-catalog-client/src/main/java/org/apache/tajo/catalog/CatalogClient.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-client/src/main/java/org/apache/tajo/catalog/CatalogClient.java b/tajo-catalog/tajo-catalog-client/src/main/java/org/apache/tajo/catalog/CatalogClient.java
index 260106f..7977b58 100644
--- a/tajo-catalog/tajo-catalog-client/src/main/java/org/apache/tajo/catalog/CatalogClient.java
+++ b/tajo-catalog/tajo-catalog-client/src/main/java/org/apache/tajo/catalog/CatalogClient.java
@@ -20,11 +20,11 @@ package org.apache.tajo.catalog;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.net.NetUtils;
import org.apache.tajo.catalog.CatalogProtocol.CatalogProtocolService.BlockingInterface;
import org.apache.tajo.conf.TajoConf;
import org.apache.tajo.conf.TajoConf.ConfVars;
import org.apache.tajo.rpc.ProtoBlockingRpcClient;
+import org.apache.tajo.util.NetUtils;
import java.io.IOException;
import java.net.InetSocketAddress;
@@ -34,7 +34,6 @@ import java.net.InetSocketAddress;
*/
public class CatalogClient extends AbstractCatalogClient {
private final Log LOG = LogFactory.getLog(CatalogClient.class);
- private final TajoConf conf;
private ProtoBlockingRpcClient client;
/**
@@ -42,24 +41,25 @@ public class CatalogClient extends AbstractCatalogClient {
*
*/
public CatalogClient(final TajoConf conf) throws IOException {
- this.conf = conf;
- connect();
+ String catalogAddr = conf.getVar(ConfVars.CATALOG_ADDRESS);
+ connect(NetUtils.createSocketAddr(catalogAddr));
}
- private void connect() throws IOException {
- String serverName = conf.getVar(ConfVars.CATALOG_ADDRESS);
- LOG.info("Trying to connect the catalog (" + serverName + ")");
- InetSocketAddress addr = NetUtils.createSocketAddr(serverName);
+ public CatalogClient(String host, int port) throws IOException {
+ connect(NetUtils.createSocketAddr(host, port));
+ }
+
+ private void connect(InetSocketAddress serverAddr) throws IOException {
+ String addrStr = NetUtils.normalizeInetSocketAddress(serverAddr);
+ LOG.info("Trying to connect the catalog (" + addrStr + ")");
try {
- client = new ProtoBlockingRpcClient(
- CatalogProtocol.class,
- addr);
+ client = new ProtoBlockingRpcClient(CatalogProtocol.class, serverAddr);
setStub((BlockingInterface) client.getStub());
} catch (Exception e) {
throw new IOException(e);
}
- LOG.info("Connected to the catalog server (" + serverName + ")");
+ LOG.info("Connected to the catalog server (" + addrStr + ")");
}
public void close() {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/MiniCatalogServer.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/MiniCatalogServer.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/MiniCatalogServer.java
index a31f3c0..79f0af9 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/MiniCatalogServer.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/MiniCatalogServer.java
@@ -30,6 +30,10 @@ public class MiniCatalogServer {
catalogServers.init(conf);
catalogServers.start();
}
+
+ public MiniCatalogServer(CatalogServer server) {
+ this.catalogServers = server;
+ }
public void shutdown() {
this.catalogServers.stop();
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/DBStore.java
----------------------------------------------------------------------
diff --git a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/DBStore.java b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/DBStore.java
index 078ca6e..8f3f960 100644
--- a/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/DBStore.java
+++ b/tajo-catalog/tajo-catalog-server/src/main/java/org/apache/tajo/catalog/store/DBStore.java
@@ -26,7 +26,10 @@ import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.tajo.catalog.*;
-import org.apache.tajo.catalog.proto.CatalogProtos.*;
+import org.apache.tajo.catalog.proto.CatalogProtos.ColumnProto;
+import org.apache.tajo.catalog.proto.CatalogProtos.IndexDescProto;
+import org.apache.tajo.catalog.proto.CatalogProtos.IndexMethod;
+import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
import org.apache.tajo.catalog.statistics.TableStat;
import org.apache.tajo.common.TajoDataTypes.Type;
import org.apache.tajo.conf.TajoConf;
@@ -124,7 +127,7 @@ public class DBStore implements CatalogStore {
Statement stmt = conn.createStatement();
ResultSet res = stmt.executeQuery(sql);
- if (res.next() == false) { // if this db version is 0
+ if (!res.next()) { // if this db version is 0
insertVersion();
return 0;
} else {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-common/src/main/java/org/apache/tajo/datum/DatumFactory.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/datum/DatumFactory.java b/tajo-common/src/main/java/org/apache/tajo/datum/DatumFactory.java
index 84bc576..4563779 100644
--- a/tajo-common/src/main/java/org/apache/tajo/datum/DatumFactory.java
+++ b/tajo-common/src/main/java/org/apache/tajo/datum/DatumFactory.java
@@ -56,6 +56,8 @@ public class DatumFactory {
return NullDatum.class;
case ARRAY:
return ArrayDatum.class;
+ case NULL:
+ return NullDatum.class;
default:
throw new UnsupportedOperationException(type.name());
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-common/src/main/java/org/apache/tajo/datum/NullDatum.java
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/java/org/apache/tajo/datum/NullDatum.java b/tajo-common/src/main/java/org/apache/tajo/datum/NullDatum.java
index 334e935..51d74e9 100644
--- a/tajo-common/src/main/java/org/apache/tajo/datum/NullDatum.java
+++ b/tajo-common/src/main/java/org/apache/tajo/datum/NullDatum.java
@@ -30,7 +30,7 @@ public class NullDatum extends Datum {
}
private NullDatum() {
- super(Type.ANY);
+ super(Type.NULL);
}
public static NullDatum get() {
@@ -64,7 +64,7 @@ public class NullDatum extends Datum {
@Override
public byte[] asByteArray() {
- return Bytes.toBytes("NULL");
+ return Bytes.toBytes("null");
}
@Override
@@ -79,7 +79,7 @@ public class NullDatum extends Datum {
@Override
public String asChars() {
- return "NULL";
+ return "null";
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-common/src/main/proto/DataTypes.proto
----------------------------------------------------------------------
diff --git a/tajo-common/src/main/proto/DataTypes.proto b/tajo-common/src/main/proto/DataTypes.proto
index 5fef1ad..f1376b8 100644
--- a/tajo-common/src/main/proto/DataTypes.proto
+++ b/tajo-common/src/main/proto/DataTypes.proto
@@ -58,6 +58,7 @@ enum Type {
BLOB = 47;
UDT = 51; // user-defined function
+ NULL = 52; // NULL type
ANY = 61;
ARRAY = 62;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/antlr4/org/apache/tajo/engine/parser/SQLLexer.g4
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/antlr4/org/apache/tajo/engine/parser/SQLLexer.g4 b/tajo-core/tajo-core-backend/src/main/antlr4/org/apache/tajo/engine/parser/SQLLexer.g4
index 3ae6ad2..2607627 100644
--- a/tajo-core/tajo-core-backend/src/main/antlr4/org/apache/tajo/engine/parser/SQLLexer.g4
+++ b/tajo-core/tajo-core-backend/src/main/antlr4/org/apache/tajo/engine/parser/SQLLexer.g4
@@ -177,6 +177,7 @@ ON : O N;
OUTER : O U T E R;
OR : O R;
ORDER : O R D E R;
+OVERWRITE : O V E R W R I T E;
PRECISION : P R E C I S I ON;
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/antlr4/org/apache/tajo/engine/parser/SQLParser.g4
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/antlr4/org/apache/tajo/engine/parser/SQLParser.g4 b/tajo-core/tajo-core-backend/src/main/antlr4/org/apache/tajo/engine/parser/SQLParser.g4
index 5955da7..5ae42c7 100644
--- a/tajo-core/tajo-core-backend/src/main/antlr4/org/apache/tajo/engine/parser/SQLParser.g4
+++ b/tajo-core/tajo-core-backend/src/main/antlr4/org/apache/tajo/engine/parser/SQLParser.g4
@@ -144,7 +144,7 @@ binary_type
===============================================================================
*/
sql
- : statement EOF
+ : statement (SEMI_COLON)? EOF
;
statement
@@ -214,7 +214,8 @@ drop_table_statement
*/
insert_statement
- : INSERT INTO table_name (LEFT_PAREN column_name_list RIGHT_PAREN)? VALUES array
+ : INSERT (OVERWRITE)? INTO table_name (LEFT_PAREN column_name_list RIGHT_PAREN)? query_expression
+ | INSERT (OVERWRITE)? INTO LOCATION path=Character_String_Literal (USING file_type=Identifier (param_clause)?)? query_expression
;
/*
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/QueryConf.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/QueryConf.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/QueryConf.java
index 34144ec..f2356a5 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/QueryConf.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/QueryConf.java
@@ -19,7 +19,6 @@
package org.apache.tajo;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
import org.apache.tajo.conf.TajoConf;
public class QueryConf extends TajoConf {
@@ -36,30 +35,4 @@ public class QueryConf extends TajoConf {
this.reloadConfiguration();
}
}
-
- public void setUser(String username) {
- setVar(ConfVars.QUERY_USERNAME, username);
- }
-
- public String getUser() {
- return getVar(ConfVars.QUERY_USERNAME);
- }
-
- public void setOutputTable(String tableName) {
- // it is determined in GlobalEngine.executeQuery().
- setVar(ConfVars.QUERY_OUTPUT_TABLE, tableName);
- }
-
- public String getOutputTable() {
- return getVar(ConfVars.QUERY_OUTPUT_TABLE);
- }
-
- public void setOutputPath(Path path) {
- // it is determined in QueryMaster.initStagingDir().
- setVar(ConfVars.QUERY_OUTPUT_DIR, path.toUri().toString());
- }
-
- public Path getOutputPath() {
- return new Path(getVar(ConfVars.QUERY_OUTPUT_DIR));
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/cli/TajoCli.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/cli/TajoCli.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/cli/TajoCli.java
index 083328e..c273f6f 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/cli/TajoCli.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/cli/TajoCli.java
@@ -349,51 +349,55 @@ public class TajoCli {
sout.println(queryId + " is killed.");
} else {
if (status.getState() == QueryState.QUERY_SUCCEEDED) {
- ResultSet res = client.getQueryResult(queryId);
- if (res == null) {
- sout.println("OK");
- return;
- }
-
- ResultSetMetaData rsmd = res.getMetaData();
- TableDesc desc = client.getResultDesc(queryId);
- sout.println("final state: " + status.getState()
- + ", init time: " + (((float)(status.getInitTime() - status.getSubmitTime())
- / 1000.0) + " sec")
- + ", execution time: " + (((float)status.getFinishTime() - status.getInitTime())
- / 1000.0) + " sec"
- + ", total response time: " + (((float)(status.getFinishTime() -
- status.getSubmitTime()) / 1000.0) + " sec"));
- sout.println("result: " + desc.getPath() + "\n");
-
- int numOfColumns = rsmd.getColumnCount();
- for (int i = 1; i <= numOfColumns; i++) {
- if (i > 1) sout.print(", ");
- String columnName = rsmd.getColumnName(i);
- sout.print(columnName);
- }
- sout.println("\n-------------------------------");
+ if (status.hasResult()) {
+ ResultSet res = client.getQueryResult(queryId);
+ if (res == null) {
+ sout.println("OK");
+ return;
+ }
- int numOfPrintedRows = 0;
- while (res.next()) {
- // TODO - to be improved to print more formatted text
+ ResultSetMetaData rsmd = res.getMetaData();
+ TableDesc desc = client.getResultDesc(queryId);
+ sout.println("final state: " + status.getState()
+ + ", init time: " + (((float)(status.getInitTime() - status.getSubmitTime())
+ / 1000.0) + " sec")
+ + ", execution time: " + (((float)status.getFinishTime() - status.getInitTime())
+ / 1000.0) + " sec"
+ + ", total response time: " + (((float)(status.getFinishTime() -
+ status.getSubmitTime()) / 1000.0) + " sec"));
+ sout.println("result: " + desc.getPath() + "\n");
+
+ int numOfColumns = rsmd.getColumnCount();
for (int i = 1; i <= numOfColumns; i++) {
if (i > 1) sout.print(", ");
- String columnValue = res.getObject(i).toString();
- sout.print(columnValue);
+ String columnName = rsmd.getColumnName(i);
+ sout.print(columnName);
}
- sout.println();
- sout.flush();
- numOfPrintedRows++;
- if (numOfPrintedRows >= PRINT_LIMIT) {
- sout.print("continue... ('q' is quit)");
- sout.flush();
- if (sin.read() == 'q') {
- break;
+ sout.println("\n-------------------------------");
+
+ int numOfPrintedRows = 0;
+ while (res.next()) {
+ // TODO - to be improved to print more formatted text
+ for (int i = 1; i <= numOfColumns; i++) {
+ if (i > 1) sout.print(", ");
+ String columnValue = res.getObject(i).toString();
+ sout.print(columnValue);
}
- numOfPrintedRows = 0;
sout.println();
+ sout.flush();
+ numOfPrintedRows++;
+ if (numOfPrintedRows >= PRINT_LIMIT) {
+ sout.print("continue... ('q' is quit)");
+ sout.flush();
+ if (sin.read() == 'q') {
+ break;
+ }
+ numOfPrintedRows = 0;
+ sout.println();
+ }
}
+ } else {
+ sout.println("OK");
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/eval/ConstEval.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/eval/ConstEval.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/eval/ConstEval.java
index d677a73..a5311c7 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/eval/ConstEval.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/eval/ConstEval.java
@@ -67,6 +67,7 @@ public class ConstEval extends EvalNode implements Comparable<ConstEval>, Clonea
case BLOB : return CatalogUtil.newDataTypesWithoutLen(TajoDataTypes.Type.BLOB);
case TEXT: return CatalogUtil.newDataTypesWithoutLen(TajoDataTypes.Type.TEXT);
case INET4: return CatalogUtil.newDataTypesWithoutLen(TajoDataTypes.Type.INET4);
+ case NULL: return CatalogUtil.newDataTypesWithoutLen(TajoDataTypes.Type.NULL);
default: return CatalogUtil.newDataTypesWithoutLen(TajoDataTypes.Type.ANY);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
index 82b1295..4a51db7 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/parser/SQLAnalyzer.java
@@ -18,6 +18,7 @@
package org.apache.tajo.engine.parser;
+import com.google.common.base.Preconditions;
import org.antlr.v4.runtime.ANTLRInputStream;
import org.antlr.v4.runtime.CommonTokenStream;
import org.antlr.v4.runtime.tree.TerminalNode;
@@ -925,6 +926,48 @@ public class SQLAnalyzer extends SQLParserBaseVisitor<Expr> {
}
@Override
+ public Expr visitInsert_statement(SQLParser.Insert_statementContext ctx) {
+ Insert insertExpr = new Insert();
+
+ if (ctx.OVERWRITE() != null) {
+ insertExpr.setOverwrite();
+ }
+
+ if (ctx.table_name() != null) {
+ insertExpr.setTableName(ctx.table_name().getText());
+
+ if (ctx.column_name_list() != null) {
+ String [] targetColumns = new String[ctx.column_name_list().Identifier().size()];
+ for (int i = 0; i < targetColumns.length; i++) {
+ targetColumns[i] = ctx.column_name_list().Identifier().get(i).getText();
+ }
+
+ insertExpr.setTargetColumns(targetColumns);
+ }
+ }
+
+ if (ctx.LOCATION() != null) {
+ insertExpr.setLocation(stripQuote(ctx.path.getText()));
+
+ if (ctx.USING() != null) {
+ insertExpr.setStorageType(ctx.file_type.getText());
+
+ if (ctx.param_clause() != null) {
+ insertExpr.setParams(getParams(ctx.param_clause()));
+ }
+ }
+ }
+
+ insertExpr.setSubQuery(visitQuery_expression(ctx.query_expression()));
+
+ Preconditions.checkState(insertExpr.hasTableName() || insertExpr.hasLocation(),
+ "Either a table name or a location should be given.");
+ Preconditions.checkState(insertExpr.hasTableName() ^ insertExpr.hasLocation(),
+ "A table name and a location cannot coexist.");
+ return insertExpr;
+ }
+
+ @Override
public Expr visitDrop_table_statement(SQLParser.Drop_table_statementContext ctx) {
return new DropTable(ctx.table_name().getText());
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BaseAlgebraVisitor.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BaseAlgebraVisitor.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BaseAlgebraVisitor.java
index 03eca32..4c06f28 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BaseAlgebraVisitor.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BaseAlgebraVisitor.java
@@ -82,6 +82,9 @@ public abstract class BaseAlgebraVisitor<T1, T2> implements AlgebraVisitor<T1, T
case Relation:
current = visitRelation(ctx, stack, (Relation) expr);
break;
+ case Insert:
+ current = visitInsert(ctx, stack, (Insert) expr);
+ break;
case CreateTable:
current = visitCreateTable(ctx, stack, (CreateTable) expr);
break;
@@ -97,6 +100,13 @@ public abstract class BaseAlgebraVisitor<T1, T2> implements AlgebraVisitor<T1, T
return current;
}
+ protected T2 visitInsert(T1 ctx, Stack<OpType> stack, Insert expr) throws PlanningException {
+ stack.push(expr.getType());
+ T2 child = visitChild(ctx, stack, expr.getSubQuery());
+ stack.pop();
+ return child;
+ }
+
@Override
public T2 visitProjection(T1 ctx, Stack<OpType> stack, Projection expr) throws PlanningException {
stack.push(expr.getType());
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BasicLogicalPlanVisitor.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BasicLogicalPlanVisitor.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BasicLogicalPlanVisitor.java
index 9bc6b7d..52d2a0a 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BasicLogicalPlanVisitor.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/BasicLogicalPlanVisitor.java
@@ -197,4 +197,12 @@ public class BasicLogicalPlanVisitor<T> implements LogicalPlanVisitor<T> {
stack.pop();
return node;
}
+
+ @Override
+ public LogicalNode visitInsert(LogicalPlan plan, InsertNode node, Stack<LogicalNode> stack, T data) throws PlanningException {
+ stack.push(node);
+ visitChild(plan, node.getSubQuery(), stack, data);
+ stack.pop();
+ return node;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/InsertNode.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/InsertNode.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/InsertNode.java
new file mode 100644
index 0000000..f85fae5
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/InsertNode.java
@@ -0,0 +1,183 @@
+/**
+ * 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.tajo.engine.planner;
+
+import com.google.gson.annotations.Expose;
+import org.apache.hadoop.fs.Path;
+import org.apache.tajo.catalog.Options;
+import org.apache.tajo.catalog.Schema;
+import org.apache.tajo.catalog.TableDesc;
+import org.apache.tajo.catalog.proto.CatalogProtos.StoreType;
+import org.apache.tajo.engine.planner.logical.LogicalNode;
+import org.apache.tajo.engine.planner.logical.LogicalNodeVisitor;
+import org.apache.tajo.engine.planner.logical.NodeType;
+import org.apache.tajo.util.TUtil;
+
+public class InsertNode extends LogicalNode implements Cloneable {
+ @Expose private boolean overwrite;
+ @Expose private TableDesc targetTableDesc;
+ @Expose private Schema targetSchema;
+ @Expose private Path path;
+ @Expose private StoreType storageType;
+ @Expose private Options options;
+ @Expose private LogicalNode subQuery;
+
+
+ public InsertNode(TableDesc desc, LogicalNode subQuery) {
+ super(NodeType.INSERT);
+ this.targetTableDesc = desc;
+ this.subQuery = subQuery;
+ this.setInSchema(subQuery.getOutSchema());
+ }
+
+ public InsertNode(Path location, LogicalNode subQuery) {
+ super(NodeType.INSERT);
+ this.path = location;
+ this.subQuery = subQuery;
+ this.setInSchema(subQuery.getOutSchema());
+ }
+
+ public boolean isOverwrite() {
+ return overwrite;
+ }
+
+ public void setOverwrite(boolean overwrite) {
+ this.overwrite = overwrite;
+ }
+
+ public boolean hasTargetTable() {
+ return this.targetTableDesc != null;
+ }
+
+ public TableDesc getTargetTable() {
+ return this.targetTableDesc;
+ }
+
+ public boolean hasTargetSchema() {
+ return this.targetSchema != null;
+ }
+
+ public Schema getTargetSchema() {
+ return this.targetSchema;
+ }
+
+ public void setTargetSchema(Schema schema) {
+ this.targetSchema = schema;
+ }
+
+ public boolean hasPath() {
+ return this.path != null;
+ }
+
+ public void setPath(Path path) {
+ this.path = path;
+ }
+
+ public Path getPath() {
+ return this.path;
+ }
+
+ public boolean hasStorageType() {
+ return this.storageType != null;
+ }
+
+ public void setStorageType(StoreType storageType) {
+ this.storageType = storageType;
+ }
+
+ public StoreType getStorageType() {
+ return this.storageType;
+ }
+
+ public boolean hasOptions() {
+ return this.options != null;
+ }
+
+ public void setOptions(Options opt) {
+ this.options = opt;
+ }
+
+ public Options getOptions() {
+ return this.options;
+ }
+
+ public LogicalNode getSubQuery() {
+ return this.subQuery;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof InsertNode) {
+ InsertNode other = (InsertNode) obj;
+ return super.equals(other)
+ && TUtil.checkEquals(this.targetTableDesc, other.targetTableDesc)
+ && TUtil.checkEquals(path, other.path)
+ && this.overwrite == other.overwrite
+ && TUtil.checkEquals(this.storageType, other.storageType)
+ && TUtil.checkEquals(options, other.options)
+ && subQuery.equals(other.subQuery);
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public Object clone() throws CloneNotSupportedException {
+ InsertNode store = (InsertNode) super.clone();
+ store.targetTableDesc = targetTableDesc != null ? targetTableDesc : null;
+ store.path = path != null ? new Path(path.toString()) : null;
+ store.overwrite = overwrite;
+ store.storageType = storageType != null ? storageType : null;
+ store.options = (Options) (options != null ? options.clone() : null);
+ store.subQuery = (LogicalNode) subQuery.clone();
+ return store;
+ }
+
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("INSERT ");
+ if (overwrite) {
+ sb.append("OVERWRITE ");
+ }
+ sb.append("INTO");
+
+ if (hasTargetTable()) {
+ sb.append(targetTableDesc);
+ }
+
+ if (hasPath()) {
+ sb.append(" LOCATION ");
+ sb.append(path);
+ }
+
+ sb.append(" [SUBQUERY]");
+
+ return sb.toString();
+ }
+
+ @Override
+ public void preOrder(LogicalNodeVisitor visitor) {
+ visitor.visit(this);
+ }
+
+ @Override
+ public void postOrder(LogicalNodeVisitor visitor) {
+ visitor.visit(this);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlan.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlan.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlan.java
index ac6cb45e..ed88cbc 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlan.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlan.java
@@ -52,6 +52,7 @@ public class LogicalPlan {
/** a map from between a block name to a block plan */
private Map<String, QueryBlock> queryBlocks = new LinkedHashMap<String, QueryBlock>();
private Map<LogicalNode, QueryBlock> queryBlockByNode = new HashMap<LogicalNode, QueryBlock>();
+ private QueryBlockGraph queryBlockGraph = new QueryBlockGraph();
private Set<LogicalNode> visited = new HashSet<LogicalNode>();
public LogicalPlan(LogicalPlanner planner) {
@@ -99,6 +100,25 @@ public class LogicalPlan {
return queryBlockByNode.get(node);
}
+ public void removeBlock(QueryBlock block) {
+ queryBlocks.remove(block.getName());
+ List<LogicalNode> tobeRemoved = new ArrayList<LogicalNode>();
+ for (Map.Entry<LogicalNode, QueryBlock> entry : queryBlockByNode.entrySet()) {
+ tobeRemoved.add(entry.getKey());
+ }
+ for (LogicalNode rn : tobeRemoved) {
+ queryBlockByNode.remove(rn);
+ }
+ }
+
+ public void connectBlocks(QueryBlock srcBlock, QueryBlock targetBlock, QueryBlockGraph.BlockType type) {
+ queryBlockGraph.connectBlocks(srcBlock.getName(), targetBlock.getName(), type);
+ }
+
+ public Collection<QueryBlockGraph.BlockEdge> getConnectedBlocks(String blockName) {
+ return queryBlockGraph.getBlockEdges(blockName);
+ }
+
public Collection<QueryBlock> getQueryBlocks() {
return queryBlocks.values();
}
@@ -112,6 +132,10 @@ public class LogicalPlan {
// if an added operator is a relation, add it to relation set.
switch (node.getType()) {
+ case STORE:
+ block.setStoreTableNode((StoreTableNode) node);
+ break;
+
case SCAN:
ScanNode relationOp = (ScanNode) node;
block.addRelation(relationOp);
@@ -120,6 +144,14 @@ public class LogicalPlan {
case GROUP_BY:
block.resolveGrouping();
break;
+
+ case SELECTION:
+ block.setSelectionNode((SelectionNode) node);
+ break;
+
+ case INSERT:
+ block.setInsertNode((InsertNode) node);
+ break;
}
@@ -311,7 +343,8 @@ public class LogicalPlan {
public class QueryBlock {
private String blockName;
- private LogicalNode blockRoot;
+ private LogicalNode rootNode;
+ private NodeType rootType;
private Map<String, ScanNode> relations = new HashMap<String, ScanNode>();
private Projection projection;
@@ -320,6 +353,8 @@ public class LogicalPlan {
private Projectable projectionNode;
private SelectionNode selectionNode;
private GroupbyNode groupingNode;
+ private StoreTableNode storeTableNode;
+ private InsertNode insertNode;
private Schema schema;
TargetListManager targetListManager;
@@ -333,16 +368,28 @@ public class LogicalPlan {
}
public boolean hasRoot() {
- return blockRoot != null;
+ return rootNode != null;
+ }
+
+ public void refresh() {
+ setRoot(rootNode);
}
public void setRoot(LogicalNode blockRoot) {
- this.blockRoot = blockRoot;
+ this.rootNode = blockRoot;
+ if (blockRoot instanceof LogicalRootNode) {
+ LogicalRootNode rootNode = (LogicalRootNode) blockRoot;
+ rootType = rootNode.getChild().getType();
+ }
queryBlockByNode.put(blockRoot, this);
}
public LogicalNode getRoot() {
- return blockRoot;
+ return rootNode;
+ }
+
+ public NodeType getRootType() {
+ return rootType;
}
public boolean containRelation(String name) {
@@ -430,6 +477,30 @@ public class LogicalPlan {
return selectionNode;
}
+ public boolean hasStoreTableNode() {
+ return this.storeTableNode != null;
+ }
+
+ public void setStoreTableNode(StoreTableNode storeTableNode) {
+ this.storeTableNode = storeTableNode;
+ }
+
+ public StoreTableNode getStoreTableNode() {
+ return this.storeTableNode;
+ }
+
+ public boolean hasInsertNode() {
+ return this.insertNode != null;
+ }
+
+ public InsertNode getInsertNode() {
+ return this.insertNode;
+ }
+
+ public void setInsertNode(InsertNode insertNode) {
+ this.insertNode = insertNode;
+ }
+
public String toString() {
return blockName;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlanVisitor.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlanVisitor.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlanVisitor.java
index 339824f..c41c534 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlanVisitor.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlanVisitor.java
@@ -47,4 +47,6 @@ public interface LogicalPlanVisitor <T> {
throws PlanningException;
LogicalNode visitStoreTable(LogicalPlan plan, StoreTableNode node, Stack<LogicalNode> stack, T data)
throws PlanningException;
+ LogicalNode visitInsert(LogicalPlan plan, InsertNode node, Stack<LogicalNode> stack, T data)
+ throws PlanningException;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlanner.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlanner.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlanner.java
index 6d4673a..361bef2 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlanner.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/LogicalPlanner.java
@@ -34,7 +34,6 @@ import org.apache.tajo.common.TajoDataTypes.DataType;
import org.apache.tajo.datum.Datum;
import org.apache.tajo.datum.DatumFactory;
import org.apache.tajo.engine.eval.*;
-import org.apache.tajo.engine.eval.EvalType;
import org.apache.tajo.engine.planner.LogicalPlan.QueryBlock;
import org.apache.tajo.engine.planner.logical.*;
import org.apache.tajo.engine.query.exception.InvalidQueryException;
@@ -84,7 +83,7 @@ public class LogicalPlanner extends BaseAlgebraVisitor<LogicalPlanner.PlanContex
* @param expr A relational algebraic expression for a query.
* @return A logical plan
*/
- public LogicalPlan createPlan(Expr expr) {
+ public LogicalPlan createPlan(Expr expr) throws PlanningException {
LogicalPlan plan = new LogicalPlan(this);
LogicalNode subroot = null;
@@ -93,11 +92,7 @@ public class LogicalPlanner extends BaseAlgebraVisitor<LogicalPlanner.PlanContex
QueryBlock rootBlock = plan.newAndGetBlock(LogicalPlan.ROOT_BLOCK);
PlanContext context = new PlanContext(plan, rootBlock);
- try {
- subroot = visitChild(context, stack, expr);
- } catch (PlanningException e) {
- e.printStackTrace();
- }
+ subroot = visitChild(context, stack, expr);
LogicalRootNode root = new LogicalRootNode();
root.setInSchema(subroot.getOutSchema());
@@ -147,7 +142,7 @@ public class LogicalPlanner extends BaseAlgebraVisitor<LogicalPlanner.PlanContex
// 2. build child plans
// 3. build scan plan
- Relation relation = (Relation) expr;
+ Relation relation = expr;
TableDesc desc = catalog.getTableDesc(relation.getName());
FromTable fromTable = new FromTable(desc);
@@ -722,6 +717,7 @@ public class LogicalPlanner extends BaseAlgebraVisitor<LogicalPlanner.PlanContex
LogicalNode subQuery = visitChild(context, stack, expr.getSubQuery());
stack.pop();
StoreTableNode storeNode = new StoreTableNode(tableName);
+ storeNode.setCreateTable();
storeNode.setChild(subQuery);
if (expr.hasTableElements()) {
@@ -825,6 +821,71 @@ public class LogicalPlanner extends BaseAlgebraVisitor<LogicalPlanner.PlanContex
return column;
}
+ protected LogicalNode visitInsert(PlanContext ctx, Stack<OpType> stack, Insert expr) throws PlanningException {
+ stack.push(expr.getType());
+ QueryBlock newQueryBlock = ctx.plan.newAnonymousBlock();
+ PlanContext newContext = new PlanContext(ctx.plan, newQueryBlock);
+ Stack<OpType> subStack = new Stack<OpType>();
+ LogicalNode subQuery = visitChild(newContext, subStack, expr.getSubQuery());
+ ctx.plan.connectBlocks(ctx.block, newQueryBlock, QueryBlockGraph.BlockType.TableSubQuery);
+ stack.pop();
+
+ InsertNode insertNode = null;
+ if (expr.hasTableName()) {
+ TableDesc desc = catalog.getTableDesc(expr.getTableName());
+ ctx.block.addRelation(new ScanNode(new FromTable(desc)));
+
+ Schema targetSchema = new Schema();
+ if (expr.hasTargetColumns()) {
+ String [] targetColumnNames = expr.getTargetColumns();
+ for (int i = 0; i < targetColumnNames.length; i++) {
+ Column targetColumn = ctx.plan.findColumn(ctx.block.getName(), new ColumnReferenceExpr(targetColumnNames[i]));
+ targetSchema.addColumn(targetColumn);
+ }
+ } else {
+ Schema targetTableSchema = desc.getMeta().getSchema();
+ for (int i = 0; i < subQuery.getOutSchema().getColumnNum(); i++) {
+ targetSchema.addColumn(targetTableSchema.getColumn(i));
+ }
+ }
+
+ checkInsertDomains(targetSchema, subQuery.getOutSchema());
+ insertNode = new InsertNode(desc, subQuery);
+ insertNode.setTargetSchema(targetSchema);
+ insertNode.setOutSchema(targetSchema);
+ }
+
+ if (expr.hasLocation()) {
+ insertNode = new InsertNode(new Path(expr.getLocation()), subQuery);
+ if (expr.hasStorageType()) {
+ insertNode.setStorageType(CatalogUtil.getStoreType(expr.getStorageType()));
+ }
+ if (expr.hasParams()) {
+ Options options = new Options();
+ options.putAll(expr.getParams());
+ insertNode.setOptions(options);
+ }
+ }
+
+ insertNode.setOverwrite(expr.isOverwrite());
+
+ return insertNode;
+ }
+
+ private static void checkInsertDomains(Schema targetTableScheme, Schema insertSchema)
+ throws PlanningException {
+ for (int i = 0; i < insertSchema.getColumnNum(); i++) {
+ if (!insertSchema.getColumn(i).getDataType().equals(targetTableScheme.getColumn(i).getDataType())) {
+ Column targetColumn = targetTableScheme.getColumn(i);
+ Column insertColumn = insertSchema.getColumn(i);
+ throw new PlanningException("ERROR: " +
+ insertColumn.getColumnName() + " is of type " + insertColumn.getDataType().getType().name() +
+ ", but target column '" + targetColumn.getColumnName() + "' is of type " +
+ targetColumn.getDataType().getType().name());
+ }
+ }
+ }
+
@Override
public LogicalNode visitDropTable(PlanContext context, Stack<OpType> stack, DropTable dropTable) {
DropTableNode dropTableNode = new DropTableNode(dropTable.getTableName());
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PlannerUtil.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PlannerUtil.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PlannerUtil.java
index 755d4cb..3f686af 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PlannerUtil.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/PlannerUtil.java
@@ -42,6 +42,10 @@ import java.util.Set;
public class PlannerUtil {
private static final Log LOG = LogFactory.getLog(PlannerUtil.class);
+ public static String normalizeTableName(String tableName) {
+ return tableName.toLowerCase();
+ }
+
public static boolean checkIfDDLPlan(LogicalNode node) {
LogicalNode baseNode = node;
if (node instanceof LogicalRootNode) {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/QueryBlockGraph.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/QueryBlockGraph.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/QueryBlockGraph.java
new file mode 100644
index 0000000..8677ba8
--- /dev/null
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/QueryBlockGraph.java
@@ -0,0 +1,91 @@
+/**
+ * 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.tajo.engine.planner;
+
+import org.apache.tajo.util.TUtil;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+
+/**
+ * It provides a graph data structure for query blocks and their relations.
+ */
+public class QueryBlockGraph {
+ private Map<String, List<BlockEdge>> blockEdges = new HashMap<String, List<BlockEdge>>();
+
+ public QueryBlockGraph() {
+ }
+
+ public int size() {
+ return blockEdges.size();
+ }
+
+ public void connectBlocks(String srcBlock, String targetBlock, BlockType type) {
+ BlockEdge newBlockEdge = new BlockEdge(targetBlock, type);
+ if (blockEdges.containsKey(srcBlock)) {
+ blockEdges.get(srcBlock).add(newBlockEdge);
+ } else {
+ blockEdges.put(srcBlock, TUtil.newList(newBlockEdge));
+ }
+ }
+
+ public Collection<BlockEdge> getBlockEdges(String srcBlock) {
+ return blockEdges.get(srcBlock);
+ }
+
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ for (Map.Entry<String, List<BlockEdge>> entry : blockEdges.entrySet()) {
+ sb.append(entry.getKey()).append("\n");
+ for (BlockEdge edge : entry.getValue()) {
+ sb.append(" |- ");
+ sb.append(edge.getTargetBlock()).append(" (").append(edge.getBlockType()).append(")");
+ }
+ sb.append("\n");
+ }
+
+ return sb.toString();
+ }
+
+ public static enum BlockType {
+ TableSubQuery,
+ ScalarSubQuery
+ }
+
+ public static class BlockEdge {
+ private String targetBlock;
+ private BlockType blockType;
+
+ public BlockEdge(String targetBlock, BlockType blockType) {
+ this.targetBlock = targetBlock;
+ this.blockType = blockType;
+ }
+
+ public String getTargetBlock() {
+ return targetBlock;
+ }
+
+ public BlockType getBlockType() {
+ return blockType;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/NodeType.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/NodeType.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/NodeType.java
index 7f1505c..35171ef 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/NodeType.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/NodeType.java
@@ -21,6 +21,8 @@
*/
package org.apache.tajo.engine.planner.logical;
+import org.apache.tajo.engine.planner.InsertNode;
+
/**
* This indicates a logical node type.
*/
@@ -31,6 +33,7 @@ public enum NodeType {
EXCEPT(ExceptNode.class),
EXPRS(EvalExprNode.class),
GROUP_BY(GroupbyNode.class),
+ INSERT(InsertNode.class),
INTERSECT(IntersectNode.class),
LIMIT(LimitNode.class),
JOIN(JoinNode.class),
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/StoreTableNode.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/StoreTableNode.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/StoreTableNode.java
index 7bdb587..f4b63c7 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/StoreTableNode.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/logical/StoreTableNode.java
@@ -35,6 +35,8 @@ public class StoreTableNode extends UnaryNode implements Cloneable {
@Expose private Column [] partitionKeys;
@Expose private boolean local;
@Expose private Options options;
+ @Expose private boolean isCreatedTable = false;
+ @Expose private boolean isOverwritten = false;
public StoreTableNode(String tableName) {
super(NodeType.STORE);
@@ -110,6 +112,22 @@ public class StoreTableNode extends UnaryNode implements Cloneable {
public Options getOptions() {
return this.options;
}
+
+ public boolean isCreatedTable() {
+ return isCreatedTable;
+ }
+
+ public void setCreateTable() {
+ isCreatedTable = true;
+ }
+
+ public boolean isOverwrite() {
+ return isOverwritten;
+ }
+
+ public void setOverwrite() {
+ isOverwritten = true;
+ }
@Override
public boolean equals(Object obj) {
@@ -120,7 +138,9 @@ public class StoreTableNode extends UnaryNode implements Cloneable {
eq = eq && this.storageType.equals(other.storageType);
eq = eq && this.numPartitions == other.numPartitions;
eq = eq && TUtil.checkEquals(partitionKeys, other.partitionKeys);
- eq = eq && TUtil.checkEquals(options, other.options);
+ eq = eq && TUtil.checkEquals(options, other.options);
+ eq = eq && isCreatedTable == other.isCreatedTable;
+ eq = eq && isOverwritten == other.isOverwritten;
eq = eq && child.equals(other.child);
return eq;
} else {
@@ -136,6 +156,8 @@ public class StoreTableNode extends UnaryNode implements Cloneable {
store.numPartitions = numPartitions;
store.partitionKeys = partitionKeys != null ? partitionKeys.clone() : null;
store.options = options != null ? (Options) options.clone() : null;
+ store.isCreatedTable = isCreatedTable;
+ store.isOverwritten = isOverwritten;
return store;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/rewrite/FilterPushDownRule.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/rewrite/FilterPushDownRule.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/rewrite/FilterPushDownRule.java
index 75d13ee..1f771e6 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/rewrite/FilterPushDownRule.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/rewrite/FilterPushDownRule.java
@@ -18,7 +18,6 @@
package org.apache.tajo.engine.planner.rewrite;
-import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import org.apache.tajo.algebra.JoinType;
import org.apache.tajo.engine.eval.EvalNode;
@@ -43,21 +42,26 @@ public class FilterPushDownRule extends BasicLogicalPlanVisitor<List<EvalNode>>
@Override
public boolean isEligible(LogicalPlan plan) {
- LogicalNode toBeOptimized = plan.getRootBlock().getRoot();
-
- return PlannerUtil.findTopNode(toBeOptimized, NodeType.SELECTION) != null;
+ for (LogicalPlan.QueryBlock block : plan.getQueryBlocks()) {
+ LogicalNode toBeOptimized = block.getRoot();
+ if (PlannerUtil.findTopNode(toBeOptimized, NodeType.SELECTION) != null) {
+ return true;
+ }
+ }
+ return false;
}
@Override
public LogicalPlan rewrite(LogicalPlan plan) throws PlanningException {
- LogicalNode root = plan.getRootBlock().getRoot();
- SelectionNode selNode = (SelectionNode) PlannerUtil.findTopNode(root, NodeType.SELECTION);
- Preconditions.checkNotNull(selNode);
-
- Stack<LogicalNode> stack = new Stack<LogicalNode>();
- EvalNode [] cnf = EvalTreeUtil.getConjNormalForm(selNode.getQual());
+ for (LogicalPlan.QueryBlock block : plan.getQueryBlocks()) {
+ if (block.hasSelectionNode()) {
+ SelectionNode filterNode = block.getSelectionNode();
+ Stack<LogicalNode> stack = new Stack<LogicalNode>();
+ EvalNode [] cnf = EvalTreeUtil.getConjNormalForm(filterNode.getQual());
+ visitChild(plan, block.getRoot(), stack, Lists.newArrayList(cnf));
+ }
+ }
- visitChild(plan, root, stack, Lists.newArrayList(cnf));
return plan;
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java
index 89a935a..4c2f9d7 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/planner/rewrite/ProjectionPushDownRule.java
@@ -49,7 +49,7 @@ public class ProjectionPushDownRule extends BasicLogicalPlanVisitor<ProjectionPu
public boolean isEligible(LogicalPlan plan) {
LogicalNode toBeOptimized = plan.getRootBlock().getRoot();
- if (PlannerUtil.checkIfDDLPlan(toBeOptimized) || !plan.getRootBlock().hasTableExpression()) {
+ if (PlannerUtil.checkIfDDLPlan(toBeOptimized) && !plan.getRootBlock().hasTableExpression()) {
LOG.info("This query skips the logical optimization step.");
return false;
}
@@ -59,40 +59,49 @@ public class ProjectionPushDownRule extends BasicLogicalPlanVisitor<ProjectionPu
@Override
public LogicalPlan rewrite(LogicalPlan plan) throws PlanningException {
- Stack<LogicalNode> stack = new Stack<LogicalNode>();
-
- PushDownContext context = new PushDownContext();
- context.plan = plan;
-
- if (plan.getRootBlock().getProjection() != null &&
- plan.getRootBlock().getProjection().isAllProjected()) {
- context.targetListManager = new TargetListManager(plan,
- plan.getRootBlock().getProjectionNode().getTargets());
- } else {
- context.targetListManager= new TargetListManager(plan, LogicalPlan.ROOT_BLOCK);
+ for (LogicalPlan.QueryBlock block : plan.getQueryBlocks()) {
+ NodeType nodeType = block.getRootType();
+ // skip a non-table-expression block.
+ if (!(nodeType == NodeType.INSERT || nodeType == NodeType.CREATE_TABLE || nodeType == NodeType.EXPRS)) {
+ Stack<LogicalNode> stack = new Stack<LogicalNode>();
+ PushDownContext context = new PushDownContext(block);
+ context.plan = plan;
+ if (block.getProjection() != null &&
+ block.getProjection().isAllProjected()) {
+ context.targetListManager = new TargetListManager(plan,
+ block.getProjectionNode().getTargets());
+ } else {
+ context.targetListManager= new TargetListManager(plan, block.getName());
+ }
+ context.upperRequired = new HashSet<Column>(block.getSchema().getColumns());
+ visitChild(plan, block.getRoot(), stack, context);
+ }
}
- context.upperRequired = new HashSet<Column>(plan.getRootBlock().getSchema().getColumns());
-
-
- visitChild(plan, plan.getRootBlock().getRoot(), stack, context);
-
return plan;
}
public static class PushDownContext {
+ LogicalPlan.QueryBlock queryBlock;
LogicalPlan plan;
TargetListManager targetListManager;
Set<Column> upperRequired;
- public PushDownContext() {
+ public PushDownContext(LogicalPlan.QueryBlock block) {
+ this.queryBlock = block;
}
public PushDownContext(ProjectionPushDownRule.PushDownContext context) {
this.plan = context.plan;
+ this.queryBlock = context.queryBlock;
this.targetListManager = context.targetListManager;
this.upperRequired = context.upperRequired;
}
+
+ public PushDownContext(ProjectionPushDownRule.PushDownContext context, LogicalPlan.QueryBlock queryBlock) {
+ this(context);
+ this.queryBlock = queryBlock;
+ }
}
@Override
@@ -117,10 +126,15 @@ public class ProjectionPushDownRule extends BasicLogicalPlanVisitor<ProjectionPu
// If all expressions are evaluated in the child operators and the last operator is projectable,
// ProjectionNode will not be necessary. It eliminates ProjectionNode.
if (context.targetListManager.isAllEvaluated() && (childNode instanceof Projectable)) {
- LogicalNode parent = stack.peek();
- // update the child node's output schemas
- child.setOutSchema(context.targetListManager.getUpdatedSchema());
- PlannerUtil.deleteNode(parent, node);
+ if (stack.isEmpty()) {
+ // update the child node's output schemas
+ child.setOutSchema(context.targetListManager.getUpdatedSchema());
+ context.queryBlock.setRoot(child);
+ } else {
+ LogicalNode parent = stack.peek();
+ child.setOutSchema(context.targetListManager.getUpdatedSchema());
+ PlannerUtil.deleteNode(parent, node);
+ }
return child;
} else {
node.setInSchema(child.getOutSchema());
@@ -351,12 +365,12 @@ public class ProjectionPushDownRule extends BasicLogicalPlanVisitor<ProjectionPu
PushDownContext context) throws PlanningException {
LogicalPlan.QueryBlock leftBlock = plan.getBlock(setNode.getLeftChild());
- PushDownContext leftContext = new PushDownContext(context);
+ PushDownContext leftContext = new PushDownContext(context, leftBlock);
leftContext.targetListManager = new TargetListManager(plan,
leftBlock.getTargetListManager().getUnEvaluatedTargets());
LogicalPlan.QueryBlock rightBlock = plan.getBlock(setNode.getRightChild());
- PushDownContext rightContext = new PushDownContext(context);
+ PushDownContext rightContext = new PushDownContext(context, rightBlock);
rightContext.targetListManager = new TargetListManager(plan,
rightBlock.getTargetListManager().getUnEvaluatedTargets());
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/query/QueryUnitRequestImpl.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/query/QueryUnitRequestImpl.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/query/QueryUnitRequestImpl.java
index cf10a4f..3697984 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/query/QueryUnitRequestImpl.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/engine/query/QueryUnitRequestImpl.java
@@ -23,6 +23,7 @@ import org.apache.tajo.ipc.TajoWorkerProtocol.Fetch;
import org.apache.tajo.ipc.TajoWorkerProtocol.QueryUnitRequestProto;
import org.apache.tajo.ipc.TajoWorkerProtocol.QueryUnitRequestProtoOrBuilder;
import org.apache.tajo.ipc.protocolrecords.QueryUnitRequest;
+import org.apache.tajo.master.QueryMeta;
import org.apache.tajo.storage.Fragment;
import java.net.URI;
@@ -40,6 +41,7 @@ public class QueryUnitRequestImpl implements QueryUnitRequest {
private Boolean interQuery;
private List<Fetch> fetches;
private Boolean shouldDie;
+ private QueryMeta queryMeta;
private QueryUnitRequestProto proto = QueryUnitRequestProto.getDefaultInstance();
private QueryUnitRequestProto.Builder builder = null;
@@ -52,10 +54,10 @@ public class QueryUnitRequestImpl implements QueryUnitRequest {
}
public QueryUnitRequestImpl(QueryUnitAttemptId id, List<Fragment> fragments,
- String outputTable, boolean clusteredOutput,
- String serializedData) {
+ String outputTable, boolean clusteredOutput,
+ String serializedData, QueryMeta queryMeta) {
this();
- this.set(id, fragments, outputTable, clusteredOutput, serializedData);
+ this.set(id, fragments, outputTable, clusteredOutput, serializedData, queryMeta);
}
public QueryUnitRequestImpl(QueryUnitRequestProto proto) {
@@ -67,13 +69,14 @@ public class QueryUnitRequestImpl implements QueryUnitRequest {
public void set(QueryUnitAttemptId id, List<Fragment> fragments,
String outputTable, boolean clusteredOutput,
- String serializedData) {
+ String serializedData, QueryMeta queryMeta) {
this.id = id;
this.fragments = fragments;
this.outputTable = outputTable;
this.clusteredOutput = clusteredOutput;
this.serializedData = serializedData;
this.isUpdated = true;
+ this.queryMeta = queryMeta;
}
@Override
@@ -178,6 +181,23 @@ public class QueryUnitRequestImpl implements QueryUnitRequest {
.setUrls(uri.toString()).build());
}
+
+ public QueryMeta getQueryMeta() {
+ QueryUnitRequestProtoOrBuilder p = viaProto ? proto : builder;
+ if (queryMeta != null) {
+ return queryMeta;
+ }
+ if (!p.hasQueryMeta()) {
+ return null;
+ }
+ this.queryMeta = new QueryMeta(p.getQueryMeta());
+ return this.queryMeta;
+ }
+
+ public void setQueryMeta(QueryMeta queryMeta) {
+ maybeInitBuilder();
+ this.queryMeta = queryMeta;
+ }
public List<Fetch> getFetches() {
initFetches();
@@ -249,6 +269,9 @@ public class QueryUnitRequestImpl implements QueryUnitRequest {
if (this.shouldDie != null) {
builder.setShouldDie(this.shouldDie);
}
+ if (this.queryMeta != null) {
+ builder.setQueryMeta(queryMeta.getProto());
+ }
}
private void mergeLocalToProto() {
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/ipc/protocolrecords/QueryUnitRequest.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/ipc/protocolrecords/QueryUnitRequest.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/ipc/protocolrecords/QueryUnitRequest.java
index c6fc632..ead5e73 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/ipc/protocolrecords/QueryUnitRequest.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/ipc/protocolrecords/QueryUnitRequest.java
@@ -24,6 +24,7 @@ package org.apache.tajo.ipc.protocolrecords;
import org.apache.tajo.QueryUnitAttemptId;
import org.apache.tajo.common.ProtoObject;
import org.apache.tajo.ipc.TajoWorkerProtocol;
+import org.apache.tajo.master.QueryMeta;
import org.apache.tajo.storage.Fragment;
import java.net.URI;
@@ -42,4 +43,5 @@ public interface QueryUnitRequest extends ProtoObject<TajoWorkerProtocol.QueryUn
public List<TajoWorkerProtocol.Fetch> getFetches();
public boolean shouldDie();
public void setShouldDie();
+ public QueryMeta getQueryMeta();
}
http://git-wip-us.apache.org/repos/asf/incubator-tajo/blob/ffdb1c71/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
----------------------------------------------------------------------
diff --git a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
index ee4b98d..4e5e5ce 100644
--- a/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
+++ b/tajo-core/tajo-core-backend/src/main/java/org/apache/tajo/master/GlobalEngine.java
@@ -29,24 +29,20 @@ import org.apache.tajo.QueryId;
import org.apache.tajo.QueryIdFactory;
import org.apache.tajo.TajoProtos;
import org.apache.tajo.algebra.Expr;
-import org.apache.tajo.catalog.CatalogService;
-import org.apache.tajo.catalog.CatalogUtil;
-import org.apache.tajo.catalog.TableDesc;
-import org.apache.tajo.catalog.TableMeta;
+import org.apache.tajo.catalog.*;
import org.apache.tajo.catalog.exception.AlreadyExistsTableException;
import org.apache.tajo.catalog.exception.NoSuchTableException;
import org.apache.tajo.catalog.statistics.TableStat;
+import org.apache.tajo.datum.NullDatum;
+import org.apache.tajo.engine.eval.ConstEval;
+import org.apache.tajo.engine.eval.FieldEval;
import org.apache.tajo.engine.exception.EmptyClusterException;
import org.apache.tajo.engine.exception.IllegalQueryStatusException;
import org.apache.tajo.engine.exception.NoSuchQueryIdException;
import org.apache.tajo.engine.exception.UnknownWorkerException;
import org.apache.tajo.engine.parser.SQLAnalyzer;
import org.apache.tajo.engine.planner.*;
-import org.apache.tajo.engine.planner.global.GlobalOptimizer;
-import org.apache.tajo.engine.planner.logical.CreateTableNode;
-import org.apache.tajo.engine.planner.logical.DropTableNode;
-import org.apache.tajo.engine.planner.logical.LogicalNode;
-import org.apache.tajo.engine.planner.logical.LogicalRootNode;
+import org.apache.tajo.engine.planner.logical.*;
import org.apache.tajo.ipc.ClientProtos;
import org.apache.tajo.master.TajoMaster.MasterContext;
import org.apache.tajo.master.querymaster.QueryInfo;
@@ -56,8 +52,12 @@ import org.apache.tajo.storage.StorageUtil;
import java.io.IOException;
import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import static org.apache.tajo.ipc.ClientProtos.GetQueryStatusResponse;
-@SuppressWarnings("unchecked")
public class GlobalEngine extends AbstractService {
/** Class Logger */
private final static Log LOG = LogFactory.getLog(GlobalEngine.class);
@@ -69,11 +69,9 @@ public class GlobalEngine extends AbstractService {
private CatalogService catalog;
private LogicalPlanner planner;
private LogicalOptimizer optimizer;
- private GlobalPlanner globalPlanner;
- private GlobalOptimizer globalOptimizer;
+ private DistributedQueryHookManager hookManager;
- public GlobalEngine(final MasterContext context)
- throws IOException {
+ public GlobalEngine(final MasterContext context) {
super(GlobalEngine.class.getName());
this.context = context;
this.catalog = context.getCatalog();
@@ -86,9 +84,10 @@ public class GlobalEngine extends AbstractService {
planner = new LogicalPlanner(context.getCatalog());
optimizer = new LogicalOptimizer();
- globalPlanner = new GlobalPlanner(context.getConf(), sm, context.getEventHandler());
+ hookManager = new DistributedQueryHookManager();
+ hookManager.addHook(new CreateTableHook());
+ hookManager.addHook(new InsertHook());
- globalOptimizer = new GlobalOptimizer();
} catch (Throwable t) {
LOG.error(t.getMessage(), t);
}
@@ -99,63 +98,70 @@ public class GlobalEngine extends AbstractService {
super.stop();
}
- public ClientProtos.GetQueryStatusResponse executeQuery(String sql)
+ public GetQueryStatusResponse executeQuery(String sql)
throws InterruptedException, IOException,
NoSuchQueryIdException, IllegalQueryStatusException,
UnknownWorkerException, EmptyClusterException {
LOG.info("SQL: " + sql);
- // parse the query
- Expr planningContext = analyzer.parse(sql);
- LogicalRootNode plan = (LogicalRootNode) createLogicalPlan(planningContext);
-
- ClientProtos.GetQueryStatusResponse.Builder responseBuilder = ClientProtos.GetQueryStatusResponse.newBuilder();
- if (PlannerUtil.checkIfDDLPlan(plan)) {
- updateQuery(plan.getChild());
+ try {
+ Expr planningContext = analyzer.parse(sql);
+ LogicalPlan plan = createLogicalPlan(planningContext);
+ LogicalRootNode rootNode = (LogicalRootNode) plan.getRootBlock().getRoot();
- responseBuilder.setQueryId(QueryIdFactory.NULL_QUERY_ID.getProto());
- responseBuilder.setResultCode(ClientProtos.ResultCode.OK);
- responseBuilder.setState(TajoProtos.QueryState.QUERY_SUCCEEDED);
- } else {
- QueryJobManager queryJobManager = context.getQueryJobManager();
- QueryInfo queryInfo = null;
- try {
- queryInfo = queryJobManager.createNewQueryJob(sql, plan);
- } catch (Exception e) {
+ GetQueryStatusResponse.Builder responseBuilder = GetQueryStatusResponse.newBuilder();
+ if (PlannerUtil.checkIfDDLPlan(rootNode)) {
+ updateQuery(rootNode.getChild());
responseBuilder.setQueryId(QueryIdFactory.NULL_QUERY_ID.getProto());
- responseBuilder.setResultCode(ClientProtos.ResultCode.ERROR);
- responseBuilder.setState(TajoProtos.QueryState.QUERY_ERROR);
- responseBuilder.setErrorMessage(StringUtils.stringifyException(e));
-
- return responseBuilder.build();
+ responseBuilder.setResultCode(ClientProtos.ResultCode.OK);
+ responseBuilder.setState(TajoProtos.QueryState.QUERY_SUCCEEDED);
+ } else {
+ QueryMeta queryMeta = new QueryMeta();
+ hookManager.doHooks(queryMeta, plan);
+
+ QueryJobManager queryJobManager = this.context.getQueryJobManager();
+ QueryInfo queryInfo;
+
+ queryInfo = queryJobManager.createNewQueryJob(queryMeta, sql, rootNode);
+
+ responseBuilder.setQueryId(queryInfo.getQueryId().getProto());
+ responseBuilder.setResultCode(ClientProtos.ResultCode.OK);
+ responseBuilder.setState(queryInfo.getQueryState());
+ if(queryInfo.getQueryMasterHost() != null) {
+ responseBuilder.setQueryMasterHost(queryInfo.getQueryMasterHost());
+ }
+ responseBuilder.setQueryMasterPort(queryInfo.getQueryMasterClientPort());
}
+ GetQueryStatusResponse response = responseBuilder.build();
- //queryJobManager.getEventHandler().handle(new QueryJobEvent(QueryJobEvent.Type.QUERY_JOB_START, queryInfo));
-
- responseBuilder.setQueryId(queryInfo.getQueryId().getProto());
- responseBuilder.setResultCode(ClientProtos.ResultCode.OK);
- responseBuilder.setState(queryInfo.getQueryState());
- if(queryInfo.getQueryMasterHost() != null) {
- responseBuilder.setQueryMasterHost(queryInfo.getQueryMasterHost());
+ return response;
+ } catch (Throwable t) {
+ LOG.error("\nStack Trace:\n" + StringUtils.stringifyException(t));
+ GetQueryStatusResponse.Builder responseBuilder = GetQueryStatusResponse.newBuilder();
+ responseBuilder.setQueryId(QueryIdFactory.NULL_QUERY_ID.getProto());
+ responseBuilder.setResultCode(ClientProtos.ResultCode.ERROR);
+ responseBuilder.setState(TajoProtos.QueryState.QUERY_ERROR);
+ String errorMessage = t.getMessage();
+ if (t.getMessage() == null) {
+ errorMessage = StringUtils.stringifyException(t);
}
- responseBuilder.setQueryMasterPort(queryInfo.getQueryMasterClientPort());
+ responseBuilder.setErrorMessage(errorMessage);
+ return responseBuilder.build();
}
-
- ClientProtos.GetQueryStatusResponse response = responseBuilder.build();
- return response;
}
- public QueryId updateQuery(String sql) throws IOException, SQLException {
+ public QueryId updateQuery(String sql) throws IOException, SQLException, PlanningException {
LOG.info("SQL: " + sql);
// parse the query
- Expr planningContext = analyzer.parse(sql);
- LogicalRootNode plan = (LogicalRootNode) createLogicalPlan(planningContext);
+ Expr expr = analyzer.parse(sql);
+ LogicalPlan plan = createLogicalPlan(expr);
+ LogicalRootNode rootNode = (LogicalRootNode) plan.getRootBlock().getRoot();
- if (!PlannerUtil.checkIfDDLPlan(plan)) {
+ if (!PlannerUtil.checkIfDDLPlan(rootNode)) {
throw new SQLException("This is not update query:\n" + sql);
} else {
- updateQuery(plan.getChild());
+ updateQuery(rootNode.getChild());
return QueryIdFactory.NULL_QUERY_ID;
}
}
@@ -177,21 +183,15 @@ public class GlobalEngine extends AbstractService {
}
}
- private LogicalNode createLogicalPlan(Expr expression) throws IOException {
+ private LogicalPlan createLogicalPlan(Expr expression) throws PlanningException {
LogicalPlan plan = planner.createPlan(expression);
- LogicalNode optimizedPlan = null;
- try {
- optimizedPlan = optimizer.optimize(plan);
- } catch (PlanningException e) {
- LOG.error(e.getMessage(), e);
- }
-
- if(LOG.isDebugEnabled()) {
+ optimizer.optimize(plan);
+ if (LOG.isDebugEnabled()) {
LOG.debug("LogicalPlan:\n" + plan.getRootBlock().getRoot());
}
- return optimizedPlan;
+ return plan;
}
private TableDesc createTable(CreateTableNode createTable) throws IOException {
@@ -272,4 +272,157 @@ public class GlobalEngine extends AbstractService {
LOG.info("Table \"" + tableName + "\" is dropped.");
}
+
+ public interface DistributedQueryHook {
+ boolean isEligible(QueryMeta queryMeta, LogicalPlan plan);
+ void hook(QueryMeta queryMeta, LogicalPlan plan) throws Exception;
+ }
+
+ public class DistributedQueryHookManager {
+ private List<DistributedQueryHook> hooks = new ArrayList<DistributedQueryHook>();
+ public void addHook(DistributedQueryHook hook) {
+ hooks.add(hook);
+ }
+
+ public void doHooks(QueryMeta queryMeta, LogicalPlan plan) {
+ for (DistributedQueryHook hook : hooks) {
+ if (hook.isEligible(queryMeta, plan)) {
+ try {
+ hook.hook(queryMeta, plan);
+ } catch (Throwable t) {
+ t.printStackTrace();
+ }
+ }
+ }
+ }
+ }
+
+ private class CreateTableHook implements DistributedQueryHook {
+
+ @Override
+ public boolean isEligible(QueryMeta queryMeta, LogicalPlan plan) {
+ if (plan.getRootBlock().hasStoreTableNode()) {
+ StoreTableNode storeTableNode = plan.getRootBlock().getStoreTableNode();
+ return storeTableNode.isCreatedTable();
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public void hook(QueryMeta queryMeta, LogicalPlan plan) throws Exception {
+ StoreTableNode storeTableNode = plan.getRootBlock().getStoreTableNode();
+ queryMeta.setOutputTable(storeTableNode.getTableName());
+ queryMeta.setCreateTable();
+ }
+ }
+
+ private class InsertHook implements DistributedQueryHook {
+
+ @Override
+ public boolean isEligible(QueryMeta queryMeta, LogicalPlan plan) {
+ return plan.getRootBlock().getRootType() == NodeType.INSERT;
+ }
+
+ @Override
+ public void hook(QueryMeta queryMeta, LogicalPlan plan) throws Exception {
+ queryMeta.setInsert();
+
+ InsertNode insertNode = plan.getRootBlock().getInsertNode();
+ StoreTableNode storeNode;
+
+ // Set QueryMeta settings, such as output table name and output path.
+ // It also remove data files if overwrite is true.
+ String outputTableName;
+ Path outputPath;
+ if (insertNode.hasTargetTable()) {
+ TableDesc desc = insertNode.getTargetTable();
+ outputTableName = desc.getName();
+ outputPath = desc.getPath();
+ } else {
+ outputTableName = PlannerUtil.normalizeTableName(insertNode.getPath().getName());
+ outputPath = insertNode.getPath();
+ queryMeta.setFileOutput();
+ }
+
+ storeNode = new StoreTableNode(outputTableName);
+ queryMeta.setOutputTable(outputTableName);
+ queryMeta.setOutputPath(outputPath);
+
+ if (insertNode.isOverwrite() && sm.exists(outputPath)) {
+ queryMeta.setOutputOverwrite();
+ storeNode.setOverwrite();
+ sm.deleteData(outputPath);
+ }
+
+ ////////////////////////////////////////////////////////////////////////////////////
+ // [TARGET TABLE] [TARGET COLUMN] [SUBQUERY Schema] /
+ // INSERT INTO TB_NAME (col1, col2) SELECT c1, c2 FROM ... /
+ ////////////////////////////////////////////////////////////////////////////////////
+ LogicalNode subQuery = insertNode.getSubQuery();
+ Schema subQueryOutSchema = subQuery.getOutSchema();
+
+ if (insertNode.hasTargetTable()) { // if a target table is given, it computes the proper schema.
+ Schema targetTableSchema = insertNode.getTargetTable().getMeta().getSchema();
+ Schema targetProjectedSchema = insertNode.getTargetSchema();
+
+ int [] targetColumnIds = new int[targetProjectedSchema.getColumnNum()];
+ int idx = 0;
+ for (Column column : targetProjectedSchema.getColumns()) {
+ targetColumnIds[idx++] = targetTableSchema.getColumnId(column.getQualifiedName());
+ }
+
+ Target [] targets = new Target[targetTableSchema.getColumnNum()];
+ boolean matched = false;
+ for (int i = 0; i < targetTableSchema.getColumnNum(); i++) {
+ Column column = targetTableSchema.getColumn(i);
+ for (int j = 0; j < targetColumnIds.length; j++) {
+ if (targetColumnIds[j] == i) {
+ Column outputColumn = subQueryOutSchema.getColumn(j);
+ targets[i] = new Target(new FieldEval(outputColumn), column.getColumnName());
+ matched = true;
+ break;
+ }
+ }
+ if (!matched) {
+ targets[i] = new Target(new ConstEval(NullDatum.get()), column.getColumnName());
+ }
+ matched = false;
+ }
+
+
+ ProjectionNode projectionNode = new ProjectionNode(targets);
+ projectionNode.setInSchema(insertNode.getSubQuery().getOutSchema());
+ projectionNode.setOutSchema(PlannerUtil.targetToSchema(targets));
+ Collection<QueryBlockGraph.BlockEdge> edges = plan.getConnectedBlocks(LogicalPlan.ROOT_BLOCK);
+ LogicalPlan.QueryBlock block = plan.getBlock(edges.iterator().next().getTargetBlock());
+ projectionNode.setChild(block.getRoot());
+
+
+ storeNode.setOutSchema(projectionNode.getOutSchema());
+ storeNode.setInSchema(projectionNode.getOutSchema());
+ storeNode.setChild(projectionNode);
+ } else {
+ storeNode.setOutSchema(subQueryOutSchema);
+ storeNode.setInSchema(subQueryOutSchema);
+ Collection<QueryBlockGraph.BlockEdge> edges = plan.getConnectedBlocks(LogicalPlan.ROOT_BLOCK);
+ LogicalPlan.QueryBlock block = plan.getBlock(edges.iterator().next().getTargetBlock());
+ storeNode.setChild(block.getRoot());
+ }
+
+ storeNode.setListPartition();
+ if (insertNode.hasStorageType()) {
+ storeNode.setStorageType(insertNode.getStorageType());
+ }
+ if (insertNode.hasOptions()) {
+ storeNode.setOptions(insertNode.getOptions());
+ }
+
+ // find a subquery query of insert node and merge root block and subquery into one query block.
+ PlannerUtil.replaceNode(plan.getRootBlock().getRoot(), storeNode, NodeType.INSERT);
+ plan.getRootBlock().refresh();
+ LogicalPlan.QueryBlock subBlock = plan.getBlock(insertNode.getSubQuery());
+ plan.removeBlock(subBlock);
+ }
+ }
}