Add Sql InputSource

[a2l007]

Add Sql InputSource support for ingesting events from RDBMS using parallel indexing.

---

This PR has:

• been self-reviewed.
• added documentation for new or modified features or behaviors.
• added unit tests or modified existing tests to cover new code paths.
• been tested in a test Druid cluster.

---

[vogievetsky]

Does this inputSource not need an inputFormat? How is that handled?

[a2l007]

SqlInputSource wouldn't require a custom inputFormat since the input data is being read off as result sets from a RDBMS source. Each sql entity is read into a local file and an iterator is returned based of this file.

[nishantmonu51]

LGTM, 👍

[pjain1]

can this be part of new module as InputSource seems to replacement for firehose related interfaces ?

[a2l007]

Thanks. Extracted it to a separate module.

[pjain1]

Thanks, however I would have just called it InputSourceModule and made it available to all processes by adding it in makeInjectorWithModules of Initialization.java similar to FirehoseModule.

[a2l007]

I can refactor the module name, but I believe this module wouldn't be needed in the broker, historical and router processes since its specific to indexing.

[pjain1]

ok

[suneet-s]

@a2l007 have you thought about how to add integration tests for something like this? Can we add innodb or something in a docker container and run it as part of the integration testing framework?

[a2l007]

@suneet-s Yeah good idea! I'm planning to add InputSource based equivalent for CombiningFirehose first as well as fix this bug: #9389
I'd take a stab at adding integration tests for both after that.

[jihoonson]

LGTM overall. Left minor comments on docs.

[jihoonson]

👍

[jihoonson]

Maybe worth mentioning one more time that these SQLs are executed in two sub tasks when you run a Parallel task.

[jihoonson]

sqls should be properly capitalized as SQLs.

[jihoonson]

Could you add a comment on why we fetch all result in local storage first? I remember this is to avoid holding database connections for too long time. It would help other developers.

[jihoonson]

nit: this doesn't have to be done in this PR, but how about making JsonIterator a CloseableIterator? It already implements Iterator and Closeable so it would be pretty simple.

[jihoonson]

Would you add more detailed docs for this parameter? It should probably mention that you have to load some extension to read from a particular type of database.

[jihoonson]

One more thing, I remember that many people from our community have been asking about how to use SqlFirehose. What do you think about adding a section that explains how to use it in production environment? To be honest, it's not clear for me what are best practices to make a scalable and efficient pipeline using this input source. For example, how do you parallelize each ingestion task (which means, how do you split queries)? How do you handle data updates in database especially after ingestion job is done? How often do you run ingestion jobs? and so on.

[a2l007]

@jihoonson I have added more details related to this in the docs. Please take a look. Since this is a one time ingestion from rdbms, updates to the source db data for a specific interval would require a new sql ingestion task be spawned for the same interval which will replace the old segments.

By the way, I would like to add a continuous ingestion from RDBMS feature as well which could use the indexing service with a RDBMS supervisor. Do you think this would be a useful feature?

[jihoonson]

I would definitely vote for having such a supervisor! That will be super useful.

[suneet-s]

Thanks for the contribution. I have some questions around how the packages are structured - should this be an extension or part of druid-core? If it should be part of core, can we move all of it under a set of smaller packages so it's easier to maintain (ie. all Sql ingestion stuff is under a 1 module, with potentially sub-packages inside it)

Also I think more logging on what to do operationally when something goes wrong would be helpful. If I'm reading the code correctly, the only real issue is running out of disk space in the temp folder - since we stream in the results. Are there any safeguards we can put in place to prevent users from blowing up the machine by issuing a SELECT * query to a very large database?

[2bethere]

Thanks for the contribution, I have a few questions.

1. If the SQL table has a timestamp like column, is there a way for me to specify this as a parameter so that not the entire table gets pulled?
2. Is there a way for me to specify which column to split this on? Because the user might already know how the table is sharded/partitioned to make it more efficient in parallel ingestion
3. If incremental loads are supported, how are duplicates handled? Do I specify a key or this is handled downstream?

[a2l007]

@2bethere
Thanks for taking a look.

If the SQL table has a timestamp like column, is there a way for me to specify this as a parameter so that not the entire table gets pulled?

You could use WHERE clauses within your SQL query to restrict the data based on your requirements. It is recommended to filter SQL queries based on the intervals specified in the granularity spec so as to avoid handling unwanted data.

Is there a way for me to specify which column to split this on? Because the user might already know how the table is sharded/partitioned to make it more efficient in parallel ingestion

There isnt a direct way to split the input data based on a column as this InputSource splits the task into sub tasks based on the the number of SQL queries. One way you could spread out the data across sub-tasks would be to introduce pagination within your SQL queries.

If incremental loads are supported, how are duplicates handled? Do I specify a key or this is handled downstream?

This InputSource is no different any other native batch InputSource types in terms of handling updates. Therefore, any changes in your source db for a specific interval would require you to ingest the entire data for that interval again and this will replace the existing segments for the interval.
Hope that helps.

[jihoonson]

Where is the interval 2013-01-01/2013-01-02 from?

[jihoonson]

I'm not sure what it means by "avoid unwanted data being retrieved and stored locally". Does this mean the subtask can modify the sql to filter out data out of the interval in the granularitySpec? Would you point me out where it is implemented?

[a2l007]

No, I meant to say that the SQL queries should have date time range based WHERE clauses with the same interval as the interval specified in the granularitySpec. If the query doesnt have date time based filters, the query would be pulling data outside the required intervals, which is unwanted. Hope it is clear now. I have added an example as well.

[jihoonson]

That makes sense 🙂

[jihoonson]

The tempFile will not be deleted if an exception is thrown in any lines above. We should catch all exceptions and delete the file properly.

[jihoonson]

Thanks for making the JsonIterator a CloseableIterator. Now you can return jsonIterator directly and remove this.

[jihoonson]

I would definitely vote for having such a supervisor! That will be super useful.

[jihoonson]

@a2l007 thanks for updating the PR. I left some more comments. Also please check the CI failures.

> mdspell --en-us --ignore-numbers --report '../docs/**/*.md'
    ../docs/ingestion/native-batch.md
     1351 | red to the other native batch InputSources, SQL InputSource behaves diff 
>> 1 spelling error found in 165 files

You may need to add InputSources to the website/.spelling file.

Another CI failure is an insufficient test coverage of the SqlEntity.

Diff coverage statistics:
------------------------------------------------------------------------------
|     lines      |    branches    |   functions    |   path
------------------------------------------------------------------------------
| 100% (2/2)     | 100% (0/0)     |  80% (4/5)     | org/apache/druid/segment/realtime/firehose/SqlFirehoseFactory.java
|  89% (25/28)   |  92% (13/14)   |  75% (24/32)   | org/apache/druid/metadata/input/SqlInputSource.java
|  75% (3/4)     | 100% (0/0)     |  80% (4/5)     | org/apache/druid/metadata/input/InputSourceModule.java
|  74% (37/50)   |  35% (7/20)    |  73% (28/38)   | org/apache/druid/metadata/input/SqlEntity.java
|  80% (4/5)     | 100% (0/0)     |  83% (5/6)     | org/apache/druid/metadata/input/SqlInputFormat.java
|  95% (21/22)   | 100% (0/0)     |  76% (19/25)   | org/apache/druid/metadata/input/SqlReader.java
------------------------------------------------------------------------------
Total diff coverage:
 - lines: 82% (92/111)
 - branches: 58% (20/34)
 - functions: 75% (84/111)
ERROR: Insufficient branch coverage of 58% (20/34). Required 65%.

You may want to add a unit test that verifies the tempFile is deleted properly even when an exception is thrown before returning CleanableFile.

[jihoonson]

Please move jg to the above try-resource clause so that i can be closed safely.

[a2l007]

@jihoonson Thanks for reviewing. I've addressed the review comments. Could you run travis again? It seems to have failed for an different reason.

[jihoonson]

I restarted the timed out test. Will take another look and finish my review soon.

[jihoonson]

LGTM. @a2l007 thanks!

[suneet-s]

Thanks for your patience with this @a2l007

Last set of concerns:

• Is there a benefit to this approach vs exporting the results of a sql query to temp files and then using another InputSource to ingest the data? I ask this because I wonder if Druid has to take on the extra maintainability burden of correctly downloading and serializing sql query results.
• There appears to be an NPE bug in CaseFoldedMap
• Better WARNING message telling users that SqlInputSource is not recommended for production use yet.

[suneet-s]

Can we mark this as a big warning in the docs? I think a similar warning should be made on line 1316 indicating that this functionality is experimental and not yet recommended for production use.

[a2l007]

Is this necessary? This inputsource has been used in a couple of production type environments internally and as long as the indexer is allocated enough disk space by the cluster operator, it shouldnt run into any issues.
I've added some text asking the user to review these points before using the input source.

[jihoonson]

I wouldn't say the SqlInputSource is an experimental. The experimental feature in Druid means either 1) the feature hasn't been stabilized yet and there could be potential bugs while using it or 2) the feature might be stable but can be removed in the future since we know there is a better way. In this case, I believe there will be a better way for ingesting from databases such as the SqlSupervisor mentioned here, but even in a better way, we can build the SqlSupervisor on top of the SqlInputSource (probably the supervisor can use it). Also, we have been providing the SqlFirehose as a non-experimental feature.

[suneet-s]

Maybe instead of experimental, we should call this feature Beta. It indicates that there is future work coming and we may want to change the behavior or configuration of the InputSource, so users shouldn't rely on the API / behavior being consistent across releases.

Since there are no integration tests, I'm concerned about calling this GA, because someone would have to run through these tests manually before each release to make sure we did not accidentally break the SqlInputSource

[jihoonson]

Maybe instead of experimental, we should call this feature Beta. It indicates that there is future work coming and we may want to change the behavior or configuration of the InputSource, so users shouldn't rely on the API / behavior being consistent across releases.

I agree that more fine-grained feature lifecycle would be nice, but it should be discussed separately in the dev mailing list. For APIs/behaviors, all Druid InputSources are UnstableApi which may change in breaking ways at any time even between minor Druid release lines.

Since there are no integration tests, I'm concerned about calling this GA, because someone would have to run through these tests manually before each release to make sure we did not accidentally break the SqlInputSource

Since all inputSources are UnstableApis, I wouldn't call any of them GA. Integration tests sound nice, we should add them in the near future.

[suneet-s]

Sorry I missed this earlier: Why did we chose json serialization? What happens if the object returned from the result set can not be serialized by the json generator?

[a2l007]

The design was made to align with one of the multiple use cases where events from json sources were imported into RDBMS and this inputsource would be used to index off of that. The data types supported by file formats would work with this Inputsource as well. Going forward I’d like to take a stab at evaluating if this result file can be hooked up to read from one of the TextReader implementations. The tricky part right now is that the intermediate row parsing logic for SqlReader is completely different compared to the TextReader one.

[suneet-s]

Since there are no integration tests, I'd like to advocate that we call this InputSource beta in the docs or something like that. Otherwise LGTM.

If others in the community are ok with the wording in the docs - I'm happy to change my vote.

@nishantmonu51 @pjain1 @jihoonson - any thoughts on this comment since you previously reviewed this change.

And thanks again @a2l007 for the PR - this will definitely make it easier for new users to try out Druid 🎉

[a2l007]

Since there are no integration tests, I'd like to advocate that we call this InputSource beta in the docs or something like that. Otherwise LGTM.

If others in the community are ok with the wording in the docs - I'm happy to change my vote.

Tagging previous reviewers: @jihoonson @nishantmonu51 @pjain 1 : Could you please provide your feedback regarding this?

[jihoonson]

I wrote my thought here. It maybe makes sense to call all InputSources Beta or something depending on what Beta means. However, we don't have such feature tags yet and so it should be discussed separately instead of in this PR.

Regarding integration tests, I would say the original author or anyone can add it later. I think it's better to not add it in this PR since we don't have a framework to easily add such a test yet, which mean the PR size can grow large. We can open an issue for those integration tests and tag the next major release version so that those issues become release blockers.

[pjain1]

We already have documentation for InputSource and its been used in production, we can just add notes there about the potential pitfalls and discuss separately what should it be called - experimental, beta or anything else.

[suneet-s]

Regarding integration tests, I would say the original author or anyone can add it later. I think it's better to not add it in this PR since we don't have a framework to easily add such a test yet, which mean the PR size can grow large. We can open an issue for those integration tests and tag the next major release version so that those issues become release blockers.

I've filed #10009 to add the integration tests - @a2l007 will you have time to add them before the 0.19 release?

@pjain1 - I see you and @jon-wei were discussing a security model for Druid in #9380 Does adding an InputSource that allows a Druid user to read any SQL table - like the metadata store, config tables, etc. break the security model you had in mind for Druid?

[suneet-s]

@pjain1 - I see you and @jon-wei were discussing a security model for Druid in #9380 Does adding an InputSource that allows a Druid user to read any SQL table - like the metadata store, config tables, etc. break the security model you had in mind for Druid?

Reading the code more closely, it looks like the user will have to provide credentials to the metadata store to read the data. This seems like a reasonable expectation. Initially I thought a user could use the credentials already configured in the Druid cluster to read the metadata store.

[suneet-s]

Approved since we have a blocking issue for 0.19 to add integration tests to verify this works

[maytasm]

Have a PR up for adding integration tests to SqlInputSource. Please see: #10080