-
Notifications
You must be signed in to change notification settings - Fork 4
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add kafka table and engine #6
base: develop
Are you sure you want to change the base?
Conversation
@@ -116,6 +117,11 @@ def __init__(self, db_name, db_url='http://localhost:8123/', | |||
self.readonly = True | |||
elif autocreate and not self.db_exists: | |||
self.create_database() | |||
self.cluster = cluster | |||
if self.cluster: | |||
self.on_cluster_sql = f'ON CLUSTER {cluster}' |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Maybe add a space just in case as having two spaces works, having 0 doesn't
self.on_cluster_sql = f'ON CLUSTER {cluster}' | |
self.on_cluster_sql = f' ON CLUSTER {cluster}' |
kafka_topic_list = '{self.topic}' | ||
kafka_group_name = '{self.group_id}' | ||
kafka_format = '{self.format}' | ||
kafka_num_consumers = {self.num_consumers}""" |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
you're missing the skip broken messages here, not sure about the schema either
@@ -487,8 +487,8 @@ def create_table_sql(cls, db): | |||
''' | |||
Returns the SQL statement for creating a table for this model. | |||
''' | |||
parts = ['CREATE TABLE IF NOT EXISTS `%s`.`%s` AS `%s`.`%s`' % (db.db_name, cls.table_name(), db.db_name, | |||
cls.engine.main_model.table_name())] | |||
parts = [f'CREATE TABLE IF NOT EXISTS `{db.db_name}`.`{cls.table_name()}` AS `{db.db_name}`.`{cls.engine.main_model.table_name()}` {db.on_cluster_sql} ('] |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why the ')' in the end?
parts = [f'CREATE TABLE IF NOT EXISTS `{db.db_name}`.`{cls.table_name()}` AS `{db.db_name}`.`{cls.engine.main_model.table_name()}` {db.on_cluster_sql} ('] | |
parts = [f'CREATE TABLE IF NOT EXISTS `{db.db_name}`.`{cls.table_name()}` AS `{db.db_name}`.`{cls.engine.main_model.table_name()}` {db.on_cluster_sql} '] |
@@ -511,7 +511,7 @@ def create_table_sql(cls, db): | |||
Returns the SQL statement for creating a table for this model. | |||
''' | |||
assert isinstance(cls.engine, Merge), "engine must be an instance of engines.Merge" | |||
parts = ['CREATE TABLE IF NOT EXISTS `%s`.`%s` (' % (db.db_name, cls.table_name())] | |||
parts = [f'CREATE TABLE IF NOT EXISTS `{db.db_name}`.`{cls.table_name()}` {db.on_cluster_sql} ('] |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ditto
Returns the SQL statement for creating a table for this model. | ||
''' | ||
assert isinstance(cls.engine, Kafka), "engine must be an instance of engines.Kafka" | ||
parts = [f'CREATE TABLE IF NOT EXISTS `{db.db_name}`.`{cls.table_name()}` {db.on_cluster_sql} ('] |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ditto
parts.append(',\n'.join(cols)) | ||
parts.append(')') |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
parts.append(',\n'.join(cols)) | |
parts.append(')') | |
parts.append('(') | |
parts.append(',\n'.join(cols)) | |
parts.append(')') |
Kafka would be proud about this metamorphosis