Repository: caikit/caikit (python)

Issue:
Support explicit extension types in DataStreamSource
## Is your feature request related to a problem? Please describe.

In the [File](https://github.com/caikit/caikit/blob/main/caikit/interfaces/common/data_model/stream_sources.py#L29) stream source message, the only field that is available is `filename`. In [DataStreamSource](https://github.com/caikit/caikit/blob/main/caikit/runtime/service_generation/data_stream_source.py#L171) the logic of how to parse the file is determined based on splitting the file's extension off.

Some clients will not have control over the file names themselves when calling `caikit`'s APIs, so the request is to allow an `extension` field (possibly `type` or `file_type`?) that will supersede the extension splitting logic, allowing users to explicitly tell `caikit` how to load the file.

## Describe alternatives you've considered

* Caikit could fall back to "try them all" if it fails to parse an extension from the field

Pull request description:
:sparkles: load files without extensions
<!--  Thanks for sending a pull request!  Here are some tips for you:
1. Make sure to read the Contributing Guide before submitting your PR: https://github.com/caikit/caikit/blob/main/CONTRIBUTING.md
2. If this PR closes another issue, add 'closes #<issue number>' somewhere in the PR summary. GitHub will automatically close that issue when this PR gets merged. Alternatively, adding 'refs #<issue number>' will not close the issue, but help provide the reviewer more context.-->

**What this PR does / why we need it**:

Closes #480 with the alternative implementation of "try everything"

**Special notes for your reviewer**:

This is naive as, but it works

**If applicable**:
- [ ] this PR contains documentation
- [ ] this PR contains unit tests
- [ ] this PR has been tested for backwards compatibility


Patch:
diff --git a/caikit/runtime/service_generation/data_stream_source.py b/caikit/runtime/service_generation/data_stream_source.py
index 1a3b933a..19964c74 100644
--- a/caikit/runtime/service_generation/data_stream_source.py
+++ b/caikit/runtime/service_generation/data_stream_source.py
@@ -167,8 +167,11 @@ class DataStreamSourceBase(DataStream):
         """Create a data stream object by deducing file extension
         and reading the file accordingly"""
 
-        full_fname = cls._get_resolved_source_path(fname)
         _, extension = os.path.splitext(fname)
+        if not extension:
+            return cls._load_from_file_without_extension(fname)
+
+        full_fname = cls._get_resolved_source_path(fname)
         log.debug3("Pulling data stream from %s file [%s]", extension, full_fname)
 
         if not fname or not os.path.isfile(full_fname):
@@ -187,6 +190,37 @@ class DataStreamSourceBase(DataStream):
             f"Extension not supported! {extension}",
         )
 
+    @classmethod
+    def _load_from_file_without_extension(cls, fname) -> DataStream:
+        """Similar to _create_data_stream_from_file, but we don't have a file extension to work
+        with. Attempt to create a data stream using one of a few well-known formats"""
+        full_fname = cls._get_resolved_source_path(fname)
+        log.debug3("Attempting to guess file type for file: %s", full_fname)
+        for factory_method in (
+            DataStream.from_json_array,
+            DataStream.from_jsonl,
+            DataStream.from_header_csv,
+        ):
+            try:
+                stream = factory_method(full_fname).map(cls._to_element_type)
+                # Iterate once and assume we have the correct file type if this works
+                stream.peek()
+                return stream
+            except Exception as e:  # pylint: disable=broad-exception-caught
+                # Catch any exception: it's hard to know which all could be thrown by any of the
+                # formatters
+                log.debug3(
+                    "Failed to load file %s using data stream factory method %s: %s",
+                    full_fname,
+                    factory_method,
+                    e,
+                    exc_info=True,
+                )
+        raise CaikitRuntimeException(
+            grpc.StatusCode.INVALID_ARGUMENT,
+            f"Could not load input file with no extension: {full_fname}",
+        )
+
     @classmethod
     def _to_element_type(cls, raw_element: Any) -> "ElementType":
         """Stream adapter to adapt from the raw json object data representations


Test patch:
diff --git a/tests/runtime/service_generation/test_data_stream_source.py b/tests/runtime/service_generation/test_data_stream_source.py
index 182b03a8..e2168d10 100644
--- a/tests/runtime/service_generation/test_data_stream_source.py
+++ b/tests/runtime/service_generation/test_data_stream_source.py
@@ -4,6 +4,7 @@ from typing import List
 import json
 import os
 import pickle
+import shutil
 import tempfile
 
 # Third Party
@@ -336,6 +337,23 @@ def test_make_data_stream_source_jsonlfile(sample_train_service, sample_jsonl_fi
     validate_data_stream(data_stream, 2, SampleTrainingType)
 
 
+def test_make_data_stream_source_from_file_with_no_extension(
+    sample_train_service, sample_json_file, sample_jsonl_file, sample_csv_file, tmp_path
+):
+    stream_type = caikit.interfaces.common.data_model.DataStreamSourceSampleTrainingType
+    for file in [sample_json_file, sample_jsonl_file, sample_csv_file]:
+        no_extension_filename, _ = os.path.splitext(os.path.join(str(tmp_path), file))
+        shutil.copyfile(file, no_extension_filename)
+
+        ds = stream_type(file=stream_type.File(filename=no_extension_filename))
+        assert isinstance(ds, DataStreamSourceBase)
+
+        data_stream = ds.to_data_stream()
+        assert isinstance(data_stream, DataStream)
+
+        validate_data_stream(data_stream, 2, SampleTrainingType)
+
+
 #################
 # List of files
 #################