AthenaLinker: Unexpected output running predict when one input is empty

[alanakilleen]

We have an recurring job that runs predict link_only in Athena. Sometimes, the incoming dataset is empty.
For version 3.9.9, that would produce an empty result set, which made sense.

After upgrading to Splink 3 version 3.9.10+, the output is unexpected. It appears that in this case the non-empty input dataset is linked to itself?

Sample Code

# Print the Splink version
import splink
print(splink.__version__)

# Set some variables
database_name = "athena_timeout_testing"
s3_bucket = "<bucket name>"
boto3_region= "us-east-2"

# Create test DataFrames
import pandas as pd
data = {'id': [0, 1, 2, 3],
        'First Name': ['Niall', 'Harry', 'Louis', 'Zayne'],
        'Last Name': ['Horan', 'Styles', 'Tomlinsin', 'Malik']}
df = pd.DataFrame(data)
df= df.astype("string")
df_empty = df.head(0)

# Create a test database in Athena
import boto3
import awswrangler as wr
boto3_session = boto3.Session(region_name=boto3_region)
wr.catalog.create_database(database_name, boto3_session=boto3_session, exist_ok=True)

# Write the Dataframes to Athena
wr.s3.to_parquet(
    boto3_session=boto3_session,
    df=df,
    path=f"s3://{s3_bucket}/test-table/",
    dataset=True,
    mode="overwrite",
    database=database_name,
    table='sample_data'
)
wr.s3.to_parquet(
    boto3_session=boto3_session,
    df=df_empty,
    path=f"s3://{s3_bucket}/test-table-empty/",
    dataset=True,
    mode="overwrite",
    database=database_name,
    table='sample_empty'
)

# Create some example settings
import splink.athena.athena_comparison_library as cl
settings = {
    "link_type": "link_only",
    "blocking_rules_to_generate_predictions": [
        "l.first_name = r.first_name and l.last_name = r.last_name",
    ],
    "comparisons": [
        cl.levenshtein_at_thresholds("first_name", [1,2], term_frequency_adjustments=True),
    ],
    "unique_id_column_name": "id",
}

# Create an AnthenaLinker and run predict (link_only) for the two Dataframes
from splink.athena.athena_linker import AthenaLinker
linker_compare_to_empty = AthenaLinker(
    input_table_or_tables=['sample_data', 'sample_empty'],
    settings_dict=settings,
    boto3_session=boto3_session,
    output_database=database_name,
    output_bucket=s3_bucket,
)
sample1_vs_empty_results = linker_compare_to_empty.predict()

# Get the output
sample1_vs_empty_results.as_record_dict()

Note that <bucket name> should be replaced with an S3 bucket.

Expected Output

[]

Actual Output

[{'match_weight': -11.361568684275182,
  'match_probability': 0.00037989362978366057,
  'source_dataset_l': 'sample_data',
  'source_dataset_r': 'sample_data',
  'id_l': '0',
  'id_r': '0',
  'first_name_l': 'Niall',
  'first_name_r': 'Niall',
  'gamma_first_name': 3,
  'last_name_l': 'Horan',
  'last_name_r': 'Horan'},
 {'match_weight': -11.361568684275182,
  'match_probability': 0.00037989362978366057,
  'source_dataset_l': 'sample_data',
  'source_dataset_r': 'sample_data',
  'id_l': '1',
  'id_r': '1',
  'first_name_l': 'Harry',
  'first_name_r': 'Harry',
  'gamma_first_name': 3,
  'last_name_l': 'Styles',
  'last_name_r': 'Styles'},
 {'match_weight': -11.361568684275182,
  'match_probability': 0.00037989362978366057,
  'source_dataset_l': 'sample_data',
  'source_dataset_r': 'sample_data',
  'id_l': '2',
  'id_r': '2',
  'first_name_l': 'Louis',
  'first_name_r': 'Louis',
  'gamma_first_name': 3,
  'last_name_l': 'Tomlinsin',
  'last_name_r': 'Tomlinsin'},
 {'match_weight': -11.361568684275182,
  'match_probability': 0.00037989362978366057,
  'source_dataset_l': 'sample_data',
  'source_dataset_r': 'sample_data',
  'id_l': '3',
  'id_r': '3',
  'first_name_l': 'Zayne',
  'first_name_r': 'Zayne',
  'gamma_first_name': 3,
  'last_name_l': 'Malik',
  'last_name_r': 'Malik'}]

[alanakilleen]

Here is a repro for Splink Version 4. I tested against 4.0.4.

Again, <bucket name> should be replaced before running the code.

Sample Code

# Print the Splink version
import splink
print(splink.__version__)

# Set some variables
database_name = "athena_timeout_testing"
s3_bucket = "<bucket name>"
boto3_region= "us-east-2"

# Create test DataFrames
import pandas as pd
data = {'id': [0, 1, 2, 3],
        'First Name': ['Niall', 'Harry', 'Louis', 'Zayne'],
        'Last Name': ['Horan', 'Styles', 'Tomlinsin', 'Malik']}
df = pd.DataFrame(data)
df= df.astype("string")
df_empty = df.head(0)

# Create a test database in Athena
import boto3
import awswrangler as wr
boto3_session = boto3.Session(region_name=boto3_region)
wr.catalog.create_database(database_name, boto3_session=boto3_session, exist_ok=True)

# Write the Dataframes to Athena
wr.s3.to_parquet(
    boto3_session=boto3_session,
    df=df,
    path=f"s3://{s3_bucket}/test-table/",
    dataset=True,
    mode="overwrite",
    database=database_name,
    table='sample_data'
)
wr.s3.to_parquet(
    boto3_session=boto3_session,
    df=df_empty,
    path=f"s3://{s3_bucket}/test-table-empty/",
    dataset=True,
    mode="overwrite",
    database=database_name,
    table='sample_empty'
)

# Create some example settings
import splink.comparison_library as cl
from splink import SettingsCreator, block_on
settings = SettingsCreator(
    link_type = "link_only",
    blocking_rules_to_generate_predictions=[
        block_on("first_name", "last_name")
    ],
    comparisons=[
        cl.LevenshteinAtThresholds("first_name", [1,2]),
    ],
    unique_id_column_name = "id"
)

# Create an AnthenaLinker and run predict (link_only) for the two Dataframes
from splink.backends.athena import AthenaAPI
db_api = AthenaAPI(
    boto3_session=boto3_session,
    output_database=database_name,
    output_bucket=s3_bucket,
)

from splink import Linker
linker_compare_to_empty = Linker([df_empty, df], settings, db_api)
sample1_vs_empty_results = linker_compare_to_empty.inference.predict()

# Get the output
sample1_vs_empty_results.as_record_dict()

Expected Output

[]

Actual Output

[{'match_weight': -3.2875681028314045,
  'match_probability': 0.09289667059784089,
  'source_dataset_l': '__splink__input_table_1',
  'source_dataset_r': '__splink__input_table_1',
  'id_l': '0',
  'id_r': '0',
  'first_name_l': 'Niall',
  'first_name_r': 'Niall',
  'gamma_first_name': 3,
  'last_name_l': 'Horan',
  'last_name_r': 'Horan'},
 {'match_weight': -3.2875681028314045,
  'match_probability': 0.09289667059784089,
  'source_dataset_l': '__splink__input_table_1',
  'source_dataset_r': '__splink__input_table_1',
  'id_l': '3',
  'id_r': '3',
  'first_name_l': 'Zayne',
  'first_name_r': 'Zayne',
  'gamma_first_name': 3,
  'last_name_l': 'Malik',
  'last_name_r': 'Malik'},
 {'match_weight': -3.2875681028314045,
  'match_probability': 0.09289667059784089,
  'source_dataset_l': '__splink__input_table_1',
  'source_dataset_r': '__splink__input_table_1',
  'id_l': '1',
  'id_r': '1',
  'first_name_l': 'Harry',
  'first_name_r': 'Harry',
  'gamma_first_name': 3,
  'last_name_l': 'Styles',
  'last_name_r': 'Styles'},
 {'match_weight': -3.2875681028314045,
  'match_probability': 0.09289667059784089,
  'source_dataset_l': '__splink__input_table_1',
  'source_dataset_r': '__splink__input_table_1',
  'id_l': '2',
  'id_r': '2',
  'first_name_l': 'Louis',
  'first_name_r': 'Louis',
  'gamma_first_name': 3,
  'last_name_l': 'Tomlinsin',
  'last_name_r': 'Tomlinsin'}]

[alanakilleen]

Not sure if this repros for other back ends, but after confirming the version 4 repro for Athena, I have created a bug for this issue: #2496.