GitHub
04/24/2026, 7:13 PMtosqltype in singersdk/typing.py
⏱️ Runtime : 9.15 milliseconds → 5.89 milliseconds (best of 177 runs)
📝 Explanation and details
I reduced repeated recursive checks over the JSON Schema by adding a single fast traversal that collects all declared types once, then performs the same decision logic using that collected set. This avoids multiple recursive scans of nested anyOf structures and reduces function-call overhead, improving runtime on complex schemas while preserving original behavior and exceptions for typical dict-like schemas.
✅ Correctness verification report:
| Test | Status |
| ----------------------------- | --------------- |
| ⚙️ Existing Unit Tests | ✅ 23 Passed |
| 🌀 Generated Regression Tests | ✅ 5413 Passed |
| ⏪ Replay Tests | 🔘 None Found |
| 🔎 Concolic Coverage Tests | 🔘 None Found |
| 📊 Tests Coverage | 100.0% |
⚙️ Click to see Existing Unit Tests
| Test File::Test Function | Original ⏱️ | Optimized ⏱️ | Speedup |
| ------------------------------------------------------------ | ----------- | ------------ | ------- |
| sql/test_typing.py::test_convert_jsonschema_type_to_sql_type | 37.8μs | 30.6μs | 23.3%✅ |
| sql/test_typing.py::test_to_sql_type | 45.4μs | 36.2μs | 25.3%✅ |
🌀 Click to see Generated Regression Tests
from future import annotations
from singer_sdk.typing import to_sql_type
def test_string_returns_varchar_type_by_default():
# A simple JSON Schema declaring a string should map to VARCHAR
schema = {"type": "string"}
codeflash_output = to_sql_type(schema)
result = codeflash_output # 4.50μs -> 3.88μs (16.1% faster)
def test_integer_returns_integer_type():
# A simple JSON Schema declaring an integer should map to INTEGER
schema = {"type": "integer"}
codeflash_output = to_sql_type(schema)
result = codeflash_output # 3.96μs -> 3.08μs (28.3% faster)
def test_number_returns_decimal_type():
# A numeric JSON Schema should map to DECIMAL
schema = {"type": "number"}
codeflash_output = to_sql_type(schema)
result = codeflash_output # 4.12μs -> 3.17μs (30.3% faster)
def test_boolean_returns_boolean_type():
# A boolean JSON Schema should map to BOOLEAN
schema = {"type": "boolean"}
codeflash_output = to_sql_type(schema)
result = codeflash_output # 4.29μs -> 2.92μs (47.1% faster)
def test_object_and_array_both_map_to_varchar():
# Both object and array types should fall back to VARCHAR per implementation
schema_obj = {"type": "object"}
schema_arr = {"type": "array"}
codeflash_output = type(to_sql_type(schema_obj)) # 2.75μs -> 2.83μs (2.93% slower)
codeflash_output = type(to_sql_type(schema_arr)) # 1.71μs -> 1.29μs (32.2% faster)
def test_unknown_or_missing_type_defaults_to_varchar():
# If no recognized type is present (including empty schema), default to VARCHAR
codeflash_output = type(to_sql_type({})) # 3.88μs -> 2.88μs (34.8% faster)
codeflash_output = type(
to_sql_type({"type": "something-not-defined"})
) # 2.58μs -> 1.50μs (72.3% faster)
def test_string_with_maxlength_sets_varchar_length():
# If a maxLength is present, the returned VARCHAR should carry that length
schema = {"type": "string", "maxLength": 5}
codeflash_output = to_sql_type(schema)
result = codeflash_output # 4.00μs -> 3.38μs (18.5% faster)
def test_type_list_and_tuple_handling():
# The implementation supports type being a list or tuple; both should be detected
schema_list = {"type": ["null", "string"]} # list form
schema_tuple = {"type": ("string",)} # tuple form
codeflash_output = type(to_sql_type(schema_list)) # 4.21μs -> 3.54μs (18.8% faster)
codeflash_output = type(
to_sql_type(schema_tuple)
) # 2.12μs -> 1.67μs (27.6% faster)
def test_anyof_nested_type_detection():
# anyOf should be recursively inspected; a nested integer should yield INTEGER
schema = {"anyOf": [{"type": "string"}, {"type": "integer"}]}
codeflash_output = to_sql_type(schema)
result = codeflash_output # 5.67μs -> 4.00μs (41.7% faster)
def test_deeply_nested_anyof_recurses_properly():
# Build nested anyOf structures to ensure recursion works (depth > 1)
nested = {"anyOf": [{"anyOf": [{"type": "boolean"}]}]}
codeflash_output = to_sql_type(nested)
result = codeflash_output # 6.62μs -> 3.42μs (93.9% faster)
def test_string_datelike_date_time_time_formats():
# Strings with datelike formats should map to date/time SQL types.
# Use common JSON Schema "format" values that the helper function recognizes.
schema_dt = {"type": "string", "format": "date-time"}
schema_t = {"type": "string", "format": "time"}
schema_d = {"type": "string", "format": "date"}
codeflash_output = to_sql_type(schema_dt)
res_dt = codeflash_output # 4.21μs -> 3.50μs (20.2% faster)
codeflash_output = to_sql_type(schema_t)
res_t = codeflash_output # 2.17μs -> 1.58μs (36.8% faster)
codeflash_output = to_sql_type(schema_d)
res_d = codeflash_output # 1.83μs -> 1.38μs (33.3% faster)
def test_string_datelike_variations_do_not_affect_unrelated():
# A string with an unrelated format should still be VARCHAR
schema = {"type": "string", "format": "email", "maxLength": 10}
codeflash_output = to_sql_type(schema)
result = codeflash_output # 3.96μs -> 3.46μs (14.5% faster)
def test_many_schema_conversions_in_loop_performance_and_consistency():
# Run 1000 conversions in a loop with predictable alternating schemas.
# This ensures function remains deterministic and performs adequately at scale.
schemas = []
for i in range(1000):
# Alternate between several schema shapes to exercise branching logic
if i % 5 == 0:
schemas.append({"type": "string"})
elif i % 5 == 1:
schemas.append({"type": "integer"})
elif i % 5 == 2:
schemas.append({"type": "number"})
elif i % 5 == 3:
schemas.append({"type": "boolean"})
else:
# include some nested anyOf and maxLength variants
schemas.append({"anyOf": [{"type": "string", "format": "date-time"}]})
# Convert all schemas and assert expected types
for idx, schema in enumerate(schemas):
codeflash_output = to_sql_type(schema)
res = codeflash_output # 1.78ms -> 1.06ms (67.1% faster)
# Decide expected class based on the same logic we used to build the schemas
if idx % 5 == 0 or idx % 5 == 1 or idx % 5 == 2 or idx % 5 == 3:
pass
def test_large_anyof_list_handling():
# Construct a large anyOf list of 1000 entries where the last entry is integer.
# This ensures the recursion/iteration over anyOf scales and finds the correct type.
anyof_list = [{"type": "string"} for _ in range(999)]
anyof_list.append({"type": "integer"})
schema = {"anyOf": anyof_list}
codeflash_output = to_sql_type(schema)
result = codeflash_output # 470μs -> 144μs (226% faster)
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
from future import annotations
from singer_sdk.typing import to_sql_type
def test_to_sql_type_object_returns_varchar():
"""Test that object type converts to VARCHAR."""
jsonschema_type = {"type": "object"}
codeflash_output = to_sql_type(jsonschema_type)
result = codeflash_output # 2.67μs -> 3.04μs (12.3% slower)
def test_to_sql_type_array_retur…
meltano/sdkGitHub
04/24/2026, 7:13 PM