Problem importing from MongoDB

[amotl]

About

Apparently, the adapter does not work well when importing data from MongoDB:

<class ‘dlt.destinations.exceptions.DatabaseTransientException’>
“_id” conflicts with system column pattern

References

• MongoDB: dlt.destinations.exceptions.DatabaseTransientException: "_id" conflicts with system column pattern cratedb-toolkit#498
• SQL DDL error InvalidColumnNameException["_foo" conflicts with system column pattern] crate#15161

[amotl]

@surister: Related to your suggestion at CTK-498 -- I'd rather fix this permanently until CRATEDB-15161 will be resolved on the dlt-cratedb side (here) when possible, to keep the workaround contained to the special adapter we are already shipping.

I've prepared the workaround for MongoDB's _id column the other day already.

dlt-cratedb/src/dlt_cratedb/impl/cratedb/utils.py

Lines 79 to 80 in 7e74d20

	# ingestr mongodb
	# "__id",

However, I wasn't able to activate it, because the current column rewrite logic is extremely dumb and just uses Python's replace() method, which would replace the string anywhere: That's too risky with an _id label.

dlt-cratedb/src/dlt_cratedb/impl/cratedb/utils.py

Lines 83 to 105 in 7e74d20

	@staticmethod
	def quirk(thing: str) -> str:
	thing = (
	thing.replace("_dlt_load_id", "__dlt_load_id")
	.replace("_dlt_id", "__dlt_id")
	.replace("_dlt_parent_id", "__dlt_parent_id")
	.replace("_dlt_list_idx", "__dlt_list_idx")
	.replace("_dlt_root_id", "__dlt_root_id")
	.replace("_kafka_msg_id", "_kafka__msg_id")
	)
	return thing
	@staticmethod
	def unquirk(thing: str) -> str:
	thing = (
	thing.replace("__dlt_load_id", "_dlt_load_id")
	.replace("__dlt_id", "_dlt_id")
	.replace("__dlt_parent_id", "_dlt_parent_id")
	.replace("__dlt_list_idx", "_dlt_list_idx")
	.replace("__dlt_root_id", "_dlt_root_id")
	.replace("_kafka__msg_id", "_kafka_msg_id")
	)
	return thing

After improving this spot to use an exact match instead, adding the workaround for MongoDB's _id column would be ready to go. If we think it will be too intrusive to "always enable" it, we can provide it selectively per designated parameter which can be propagated from ingestr's database address URL, when applicable. ¹

Footnotes

1. See also the patch about enhancing ingestr's Kafka nozzle with additional features and parameters to control them, per https://github.com/bruin-data/ingestr/pull/306. ↩