Add support for druid data ingestion

[beinan]

• Add support for 'insert into select'
• Add support for CTAS
• Add support for native batch (parallel) ingestion
  • Ingest data from a local holder

Ingest by CTAS
create table new_dataset7 as select localtimestamp as __time, 'beinan' as name, bigint '18' as age, double '188.88' as height;

presto:druid> desc new_dataset7;
 Column |   Type    | Extra | Comment
--------+-----------+-------+---------
 __time | timestamp |       |
 age    | bigint    |       |
 height | double    |       |
 name   | varchar   |       |

Ingest by Insert:
insert into new_dataset7 select localtimestamp as __time, bigint '18' as age, double '288.88' as height, "Mike" as name;
or even more
insert into new_dataset7 SELECT cast(date_column as timestamp),bigint '18' as age, double '188.88' as height, 'aaa' as name FROM (VALUES (SEQUENCE(FROM_ISO8601_DATE('2010-01-01'), FROM_ISO8601_DATE('2020-12-31'), INTERVAL '1' DAY) ) ) AS t1(date_array) CROSS JOIN UNNEST(date_array) AS t2(date_column);

Limitation:
Currently only supports limited data types such as timestamp, varchar, bigint, double and float.

== RELEASE NOTES ==

Druid Changes
* Add support for data ingestion

[zhenxiao]

nice work, @beinan
LGTM, only 2 minor issues

[zhenxiao]

hi @beinan is it possible to add insert into and CTAS tests in: TestDruidQueryGenerator?

[beinan]

hi @beinan is it possible to add insert into and CTAS tests in: TestDruidQueryGenerator?

Hey @zhenxiao I tried to make some unit test in TestDruidQueryGenerator. But looks like TableWriteNode is an InternalPlanNode, which cannot be passed into connector's optimizer. I got an exception as below

`java.lang.IllegalArgumentException: PlanVisitor is only for connector to use; InternalPlanNode should never use it

at com.google.common.base.Preconditions.checkArgument(Preconditions.java:141)
at com.facebook.presto.sql.planner.plan.InternalPlanNode.accept(InternalPlanNode.java:32)
at com.facebook.presto.druid.DruidQueryGenerator.generate(DruidQueryGenerator.java:117)
at com.facebook.presto.druid.TestDruidQueryGenerator.testDQL(TestDruidQueryGenerator.java:52)
at com.facebook.presto.druid.TestDruidQueryGenerator.testDQL(TestDruidQueryGenerator.java:42)
at com.facebook.presto.druid.TestDruidQueryGenerator.testDQL(TestDruidQueryGenerator.java:64)
at com.facebook.presto.druid.TestDruidQueryGenerator.testDQL(TestDruidQueryGenerator.java:69)
at com.facebook.presto.druid.TestDruidQueryGenerator.testDQL(TestDruidQueryGenerator.java:74)
at com.facebook.presto.druid.TestDruidQueryGenerator.testTableWriteFromSelect(TestDruidQueryGenerator.java:205)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:104)
at org.testng.internal.Invoker.invokeMethod(Invoker.java:645)
at org.testng.internal.Invoker.invokeTestMethod(Invoker.java:851)
at org.testng.internal.Invoker.invokeTestMethods(Invoker.java:1177)
at org.testng.internal.TestMethodWorker.invokeTestMethods(TestMethodWorker.java:129)
at org.testng.internal.TestMethodWorker.run(TestMethodWorker.java:112)
at org.testng.TestRunner.privateRun(TestRunner.java:756)
at org.testng.TestRunner.run(TestRunner.java:610)
at org.testng.SuiteRunner.runTest(SuiteRunner.java:387)
at org.testng.SuiteRunner.runSequentially(SuiteRunner.java:382)
at org.testng.SuiteRunner.privateRun(SuiteRunner.java:340)
at org.testng.SuiteRunner.run(SuiteRunner.java:289)
at org.testng.SuiteRunnerWorker.runSuite(SuiteRunnerWorker.java:52)
at org.testng.SuiteRunnerWorker.run(SuiteRunnerWorker.java:86)
at org.testng.TestNG.runSuitesSequentially(TestNG.java:1293)
at org.testng.TestNG.runSuitesLocally(TestNG.java:1218)
at org.testng.TestNG.runSuites(TestNG.java:1133)
at org.testng.TestNG.run(TestNG.java:1104)
at com.intellij.rt.testng.IDEARemoteTestNG.run(IDEARemoteTestNG.java:66)
at com.intellij.rt.testng.RemoteTestNGStarter.main(RemoteTestNGStarter.java:110)

`

Is there any other way we can test insert into and ctas cases?

[zhenxiao]

get it. I think we might need to add end to end integration test for Druid.
@beinan could you please file an issue for druid connector end to end test? especially CTAS and insert test
I will merge this PR

[beinan]

get it. I think we might need to add end to end integration test for Druid.
@beinan could you please file an issue for druid connector end to end test? especially CTAS and insert test
I will merge this PR

issue #15029 created.