feat: Add explicit support for flattening features (#103)

.github/workflows/meltano-run.yml

@@ -20,6 +20,10 @@ jobs:
           mapping: whitelist
           output_db: tap_smoke_test.db
           target_table: animals
+        - tap: people
+          mapping: flatten
+          output_db: people.db
+          target_table: people
     steps:
     - uses: actions/checkout@v4
 

README.md

@@ -12,7 +12,37 @@ This mapper plugin is fully compliant with the Singer Spec and can be placed in
 
 For a collection of examples, take a look at [examples/README.md](examples/README.md).
 
-Here's one of them!:
+## Settings
+
+| Setting             | Required | Default | Description |
+|:--------------------|:--------:|:-------:|:------------|
+| stream_maps         | True     | None    | Stream maps |
+| flattening_enabled  | False    | None    | 'True' to enable schema flattening and automatically expand nested properties. |
+| flattening_max_depth| False    | None    | The max depth to flatten schemas. |
+| stream_map_config   | False    | None    | User-defined config values to be used within map expressions. |
+
+A full list of supported settings and capabilities is available by running: `meltano-map-transformer --about`
+
+## Installation
+
+We recommend using GitHub release tags when installing with `pip`. We also recommend using `pipx` or `meltano` instead of installing with `pip` directly.
+
+You can see a full list of published releases [here](https://github.com/MeltanoLabs/meltano-map-transform/releases).
+
+For example:
+
+```
+# Update the below with the latest published release:
+#  https://github.com/MeltanoLabs/meltano-map-transform/releases
+
+# Install with pip, ideally in a new virtual environment to avoid conflicts:
+pip install git+https://github.com/MeltanoLabs/[email protected]
+
+# Or better yet, use `pipx` so that virtual environments are managed automatically:
+pipx install git+https://github.com/MeltanoLabs/[email protected]
+```
+
+### Meltano installation instructions
 
 1. Add this plugin to your Meltano project
 

fixtures/people.singer

@@ -0,0 +1,9 @@
+{"type": "SCHEMA", "stream": "people", "schema": {"required": ["id"], "type": "object", "properties": {"id": {"type": "integer"}, "name": {"type": "string"}, "address": {"type": "object", "properties": {"street": {"type": "string"}, "city": {"type": "string"}, "state": {"type": "string"}, "zip": {"type": "string"}, "coordinates": {"type": "object", "properties": {"latitude": {"type": "number"}, "longitude": {"type": "number"}}}}}}}}
+{"type": "RECORD", "stream": "people", "record": {"id": 1, "name": "Jane Doe", "address": {"street": "456 Elm St", "city": "Smallville", "state": "NY", "zip": "67890", "coordinates": {"latitude": 40.758896, "longitude": -73.985130}}}}
+{"type": "RECORD", "stream": "people", "record": {"id": 2, "name": "John Smith", "address": {"street": "789 Oak Ave", "city": "Metropolis", "state": "IL", "zip": "54321", "coordinates": {"latitude": 41.878114, "longitude": -87.629799}}}}
+{"type": "RECORD", "stream": "people", "record": {"id": 3, "name": "Jane Smith", "address": {"street": "246 Maple St", "city": "Gotham", "state": "NY", "zip": "11111", "coordinates": {"latitude": 40.712776, "longitude": -74.005974}}}}
+{"type": "RECORD", "stream": "people", "record": {"id": 4, "name": "John Doe", "address": {"street": "369 Pine Rd", "city": "Central City", "state": "CA", "zip": "22222", "coordinates": {"latitude": 37.7749, "longitude": -122.4194}}}}
+{"type": "RECORD", "stream": "people", "record": {"id": 5, "name": "Jane Wilson", "address": {"street": "159 Cedar Blvd", "city": "Star City", "state": "IL", "zip": "33333", "coordinates": {"latitude": 41.872389, "longitude": -87.650047}}}}
+{"type": "RECORD", "stream": "people", "record": {"id": 6, "name": "John Wilson", "address": {"street": "753 Maple Ave", "city": "Coast City", "state": "CA", "zip": "44444", "coordinates": {"latitude": 34.052235, "longitude": -118.243683}}}}
+{"type": "RECORD", "stream": "people", "record": {"id": 7, "name": "Jane Johnson", "address": {"street": "369 Oak St", "city": "Gotham", "state": "NY", "zip": "55555", "coordinates": {"latitude": 40.712776, "longitude": -74.005974}}}}
+{"type": "RECORD", "stream": "people", "record": {"id": 8, "name": "John Johnson", "address": {"street": "159 Elm Rd", "city": "Central City", "state": "CA", "zip": "66666", "coordinates": {"latitude": 37.7749, "longitude": -122.4194}}}}

meltano.yml

@@ -20,6 +20,9 @@ plugins:
       kind: integer
     - name: streams
       kind: array
+  - name: people
+    namespace: people
+    executable: scripts/people.sh
   loaders:
   - name: target-sqlite
     variant: meltanolabs
@@ -55,6 +58,11 @@ plugins:
             id: id
             description: description
             __else__: __NULL__
+    - name: flatten
+      config:
+        stream_maps: {}
+        flattening_enabled: true
+        flattening_max_depth: 1
 environments:
 - name: dev
   config:

meltano_map_transform/mapper.py

@@ -45,6 +45,19 @@ class StreamTransform(InlineMapper):
             required=True,
             description="Stream maps",
         ),
+        th.Property(
+            "flattening_enabled",
+            th.BooleanType(),
+            description=(
+                "'True' to enable schema flattening and automatically expand nested "
+                "properties."
+            ),
+        ),
+        th.Property(
+            "flattening_max_depth",
+            th.IntegerType(),
+            description="The max depth to flatten schemas.",
+        ),
     ).to_dict()
 
     def __init__(
@@ -93,13 +106,12 @@ def map_schema_message(
             message_dict.get("key_properties", []),
         )
         for stream_map in self.mapper.stream_maps[stream_id]:
-            schema_message = singer.SchemaMessage(
+            yield singer.SchemaMessage(
                 stream_map.stream_alias,
                 stream_map.transformed_schema,
                 stream_map.transformed_key_properties,
                 message_dict.get("bookmark_keys", []),
             )
-            yield schema_message
 
     def map_record_message(
         self,
@@ -119,13 +131,12 @@ def map_record_message(
         for stream_map in self.mapper.stream_maps[stream_id]:
             mapped_record = stream_map.transform(message_dict["record"])
             if mapped_record is not None:
-                record_message = singer.RecordMessage(
+                yield singer.RecordMessage(
                     stream=stream_map.stream_alias,
                     record=mapped_record,
                     version=message_dict.get("version"),
                     time_extracted=utc_now(),
                 )
-                yield record_message
 
     def map_state_message(self, message_dict: dict) -> list[singer.Message]:
         """Do nothing to the message.

plugins/extractors/tap-smoke-test.lock

@@ -0,0 +1,22 @@
+{
+  "plugin_type": "extractors",
+  "name": "tap-smoke-test",
+  "namespace": "tap_smoke_test",
+  "label": "tap-smoke-test",
+  "pip_url": "git+https://github.com/meltano/tap-smoke-test.git",
+  "executable": "tap-smoke-test",
+  "capabilities": [
+    "discover",
+    "catalog"
+  ],
+  "settings": [
+    {
+      "name": "schema_inference_record_count",
+      "kind": "integer"
+    },
+    {
+      "name": "streams",
+      "kind": "array"
+    }
+  ]
+}

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "meltano-map-transform"
-version = "0.1.0"
+version = "0.4.1"
 description = "`meltano-map-transform` is a Singer-compatible inline map transformer, built with the Meltano SDK for Singer Taps."
 authors = [
   "Meltano Team and Contributors",
@@ -10,15 +10,18 @@ keywords = [
   "singer.io",
   "Inline Transformation",
 ]
-license = "Apache 2.0"
+license = "Apache-2.0"
 readme = "README.md"
 homepage = "https://github.com/MeltanoLabs/meltano-map-transform"
 repository = "https://github.com/MeltanoLabs/meltano-map-transform"
 documentation = "https://github.com/MeltanoLabs/meltano-map-transform#readme"
 
 [tool.poetry.dependencies]
 python = "<3.12,>=3.7.1"
-singer-sdk = "~=0.31.0"
+
+[tool.poetry.dependencies.singer-sdk]
+allow-prereleases = true
+version = "~=0.32.0b1"
 
 [tool.poetry.dev-dependencies]
 pytest = "^7.4.2"

scripts/people.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+cat fixtures/people.singer